Most of the new ipcluster is now working, including a nice client.

IPython/config/profile/ipython_config_cluster.py

@@ -0,0 +1,17 b''
+c = get_config()
+
+# This can be used at any point in a config file to load a sub config
+# and merge it into the current one.
+load_subconfig('ipython_config.py')
+
+lines = """
+from IPython.kernel.client import *
+"""
+
+# You have to make sure that attributes that are containers already
+# exist before using them.  Simple assigning a new list will override
+# all previous values.
+if hasattr(c.Global, 'exec_lines'):
+    c.Global.exec_lines.append(lines)
+else:
+    c.Global.exec_lines = [lines]
  No newline at end of file

IPython/config/default/ipcluster_config.py

@@ -16,6 +16,7 b' c = get_config()'
 # c.Global.log_to_file = False
 # c.Global.n = 2
 # c.Global.reset_config = False
+# c.Global.clean_logs = True
 
 # c.MPIExecLauncher.mpi_cmd = ['mpiexec']
 # c.MPIExecLauncher.mpi_args = []

IPython/config/default/ipcontroller_config.py

@@ -8,15 +8,11 b' c = get_config()'
 
 # Basic Global config attributes
 # c.Global.log_to_file = False
+# c.Global.clean_logs = True
 # c.Global.import_statements = ['import math']
 # c.Global.reuse_furls = True
 # c.Global.secure = True
 
-# You shouldn't have to modify these
-# c.Global.log_dir_name = 'log'
-# c.Global.security_dir_name = 'security'
-
-
 #-----------------------------------------------------------------------------
 # Configure the client services
 #-----------------------------------------------------------------------------

IPython/config/default/ipengine_config.py

@@ -1,13 +1,17 b''
 c = get_config()
 
 # c.Global.log_to_file = False
+# c.Global.clean_logs = False
 # c.Global.exec_lines = ['import numpy']
-# c.Global.log_dir_name = 'log'
-# c.Global.security_dir_name = 'security'
 # c.Global.log_level = 10
 # c.Global.shell_class = 'IPython.kernel.core.interpreter.Interpreter'
 # c.Global.furl_file_name = 'ipcontroller-engine.furl'
 # c.Global.furl_file = ''
+# The max number of connection attemps and the initial delay between
+# those attemps.
+# c.Global.connect_delay = 0.1
+# c.Global.connect_max_tries = 15
+
 
 # c.MPI.use = ''
 # c.MPI.mpi4py = """from mpi4py import MPI as mpi

IPython/kernel/asyncclient.py

@@ -1,3 +1,4 b''
+#!/usr/bin/env python
 # encoding: utf-8
 
 """Asynchronous clients for the IPython controller.
@@ -9,32 +10,32 b' deferreds to the result.'
 
 The main methods are are `get_*_client` and `get_client`.
 """
-
-__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
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 
 from IPython.kernel import codeutil
-from IPython.kernel.clientconnector import ClientConnector
+from IPython.kernel.clientconnector import (
+    AsyncClientConnector,
+    AsyncCluster
+)
 
 # Other things that the user will need
 from IPython.kernel.task import MapTask, StringTask
 from IPython.kernel.error import CompositeError
 
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 # Code
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 
-_client_tub = ClientConnector()
+_client_tub = AsyncClientConnector()
 get_multiengine_client = _client_tub.get_multiengine_client
 get_task_client = _client_tub.get_task_client
 get_client = _client_tub.get_client

IPython/kernel/client.py

@@ -1,3 +1,4 b''
+#!/usr/bin/env python
 # encoding: utf-8
 
 """This module contains blocking clients for the controller interfaces.
@@ -15,27 +16,30 b' The main classes in this module are:'
     * CompositeError
 """
 
-__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
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 
+from cStringIO import StringIO
 import sys
+import warnings
 
 # from IPython.utils import growl
 # growl.start("IPython1 Client")
 
 
 from twisted.internet import reactor
-from IPython.kernel.clientconnector import ClientConnector
+from twisted.internet.error import PotentialZombieWarning
+from twisted.python import log
+
+from IPython.kernel.clientconnector import ClientConnector, Cluster
 from IPython.kernel.twistedutil import ReactorInThread
 from IPython.kernel.twistedutil import blockingCallFromThread
 
@@ -51,46 +55,33 b' from IPython.kernel.error import CompositeError'
 # Code
 #-------------------------------------------------------------------------------
 
-_client_tub = ClientConnector()
-
-
-def get_multiengine_client(furl_or_file=''):
-    """Get the blocking MultiEngine client.
-    
-    :Parameters:
-        furl_or_file : str
-            A furl or a filename containing a furl.  If empty, the
-            default furl_file will be used
-            
-    :Returns:
-        The connected MultiEngineClient instance
-    """
-    client = blockingCallFromThread(_client_tub.get_multiengine_client, 
-        furl_or_file)
-    return client.adapt_to_blocking_client()
-
-def get_task_client(furl_or_file=''):
-    """Get the blocking Task client.
-    
-    :Parameters:
-        furl_or_file : str
-            A furl or a filename containing a furl.  If empty, the
-            default furl_file will be used
-            
-    :Returns:
-        The connected TaskClient instance
-    """
-    client = blockingCallFromThread(_client_tub.get_task_client, 
-        furl_or_file)
-    return client.adapt_to_blocking_client()
+warnings.simplefilter('ignore', PotentialZombieWarning)
 
+_client_tub = ClientConnector()
 
+get_multiengine_client = _client_tub.get_multiengine_client
+get_task_client = _client_tub.get_task_client
 MultiEngineClient = get_multiengine_client
 TaskClient = get_task_client
 
-
+twisted_log = StringIO()
+log.startLogging(sys.stdout, setStdout=0)
 
 # Now we start the reactor in a thread
 rit = ReactorInThread()
 rit.setDaemon(True)
-rit.start()
  No newline at end of file
+rit.start()
+
+
+
+
+__all__ = [
+    'MapTask',
+    'StringTask',
+    'MultiEngineClient',
+    'TaskClient',
+    'CompositeError',
+    'get_task_client',
+    'get_multiengine_client',
+    'Cluster'
+]

IPython/kernel/clientconnector.py

@@ -1,142 +1,229 b''
+#!/usr/bin/env python
 # encoding: utf-8
 
-"""A class for handling client connections to the controller."""
+"""Facilities for handling client connections to the controller."""
 
-__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
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 
-from twisted.internet import defer
+import os
 
-from IPython.kernel.fcutil import Tub, UnauthenticatedTub
+from IPython.kernel.fcutil import Tub, find_furl
+from IPython.kernel.clusterdir import ClusterDir, ClusterDirError
+from IPython.kernel.launcher import IPClusterLauncher
+from IPython.kernel.twistedutil import gatherBoth, make_deferred
+from IPython.kernel.twistedutil import blockingCallFromThread
 
-from IPython.kernel.config import config_manager as kernel_config_manager
 from IPython.utils.importstring import import_item
-from IPython.kernel.fcutil import find_furl
+from IPython.utils.genutils import get_ipython_dir
 
-co = kernel_config_manager.get_config_obj()
-client_co = co['client']
+from twisted.internet import defer
+from twisted.python import failure
 
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 # The ClientConnector class
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 
-class ClientConnector(object):
-    """
-    This class gets remote references from furls and returns the wrapped clients.
-    
-    This class is also used in `client.py` and `asyncclient.py` to create 
-    a single per client-process Tub.
+
+class AsyncClientConnector(object):
+    """A class for getting remote references and clients from furls.
+
+    This start a single :class:`Tub` for all remote reference and caches
+    references.
     """
-    
+
     def __init__(self):
         self._remote_refs = {}
         self.tub = Tub()
         self.tub.startService()
-    
-    def get_reference(self, furl_or_file):
+
+    def _find_furl(self, profile='default', cluster_dir=None, 
+                   furl_or_file=None, furl_file_name=None,
+                   ipythondir=None):
+        """Find a FURL file by profile+ipythondir or cluster dir.
+
+        This raises an exception if a FURL file can't be found.
         """
-        Get a remote reference using a furl or a file containing a furl.
-        
+        # Try by furl_or_file
+        if furl_or_file is not None:
+            try:
+                furl = find_furl(furl_or_file)
+            except ValueError:
+                return furl
+
+        if furl_file_name is None:
+            raise ValueError('A furl_file_name must be provided')
+
+        # Try by cluster_dir
+        if cluster_dir is not None:
+            cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
+            sdir = cluster_dir_obj.security_dir
+            furl_file = os.path.join(sdir, furl_file_name)
+            return find_furl(furl_file)
+
+        # Try by profile
+        if ipythondir is None:
+            ipythondir = get_ipython_dir()
+        if profile is not None:
+            cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
+                ipythondir, profile)
+            sdir = cluster_dir_obj.security_dir
+            furl_file = os.path.join(sdir, furl_file_name)
+            return find_furl(furl_file)
+
+        raise ValueError('Could not find a valid FURL file.')
+
+    def get_reference(self, furl_or_file):
+        """Get a remote reference using a furl or a file containing a furl.
+
         Remote references are cached locally so once a remote reference
         has been retrieved for a given furl, the cached version is 
         returned.
-        
-        :Parameters:
-            furl_or_file : str
-                A furl or a filename containing a furl
-        
-        :Returns:
-            A deferred to a remote reference
+
+        Parameters
+        ----------
+        furl_or_file : str
+            A furl or a filename containing a furl
+
+        Returns
+        -------
+        A deferred to a remote reference
         """
         furl = find_furl(furl_or_file)
         if furl in self._remote_refs:
             d = defer.succeed(self._remote_refs[furl])
         else:
             d = self.tub.getReference(furl)
-            d.addCallback(self.save_ref, furl)
+            d.addCallback(self._save_ref, furl)
         return d
         
-    def save_ref(self, ref, furl):
-        """
-        Cache a remote reference by its furl.
-        """
+    def _save_ref(self, ref, furl):
+        """Cache a remote reference by its furl."""
         self._remote_refs[furl] = ref
         return ref
         
-    def get_task_client(self, furl_or_file=''):
-        """
-        Get the task controller client.
+    def get_task_client(self, profile='default', cluster_dir=None,
+                        furl_or_file=None, ipythondir=None):
+        """Get the task controller client.
         
-        This method is a simple wrapper around `get_client` that allow
-        `furl_or_file` to be empty, in which case, the furls is taken
-        from the default furl file given in the configuration.
+        This method is a simple wrapper around `get_client` that passes in
+        the default name of the task client FURL file.  Usually only
+        the ``profile`` option will be needed.  If a FURL file can't be
+        found by its profile, use ``cluster_dir`` or ``furl_or_file``.
         
-        :Parameters:
-            furl_or_file : str
-                A furl or a filename containing a furl.  If empty, the
-                default furl_file will be used
-                
-        :Returns:
-            A deferred to the actual client class
-        """
-        task_co = client_co['client_interfaces']['task']
-        if furl_or_file:
-            ff = furl_or_file
-        else:
-            ff = task_co['furl_file']
-        return self.get_client(ff)
+        Parameters
+        ----------
+        profile : str
+            The name of a cluster directory profile (default="default"). The
+            cluster directory "cluster_<profile>" will be searched for
+            in ``os.getcwd()``, the ipythondir and then in the directories
+            listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
+        cluster_dir : str
+            The full path to a cluster directory.  This is useful if profiles
+            are not being used.
+        furl_or_file : str
+            A furl or a filename containing a FURLK. This is useful if you 
+            simply know the location of the FURL file.
+        ipythondir : str
+            The location of the ipythondir if different from the default.
+            This is used if the cluster directory is being found by profile.
 
-    def get_multiengine_client(self, furl_or_file=''):
+        Returns
+        -------
+        A deferred to the actual client class.
         """
-        Get the multiengine controller client.
+        return self.get_client(
+            profile, cluster_dir, furl_or_file, 
+            'ipcontroller-tc.furl', ipythondir
+        )
+
+    def get_multiengine_client(self, profile='default', cluster_dir=None,
+                               furl_or_file=None, ipythondir=None):
+        """Get the multiengine controller client.
         
-        This method is a simple wrapper around `get_client` that allow
-        `furl_or_file` to be empty, in which case, the furls is taken
-        from the default furl file given in the configuration.
+        This method is a simple wrapper around `get_client` that passes in
+        the default name of the task client FURL file.  Usually only
+        the ``profile`` option will be needed.  If a FURL file can't be
+        found by its profile, use ``cluster_dir`` or ``furl_or_file``.
         
-        :Parameters:
-            furl_or_file : str
-                A furl or a filename containing a furl.  If empty, the
-                default furl_file will be used
+        Parameters
+        ----------
+        profile : str
+            The name of a cluster directory profile (default="default"). The
+            cluster directory "cluster_<profile>" will be searched for
+            in ``os.getcwd()``, the ipythondir and then in the directories
+            listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
+        cluster_dir : str
+            The full path to a cluster directory.  This is useful if profiles
+            are not being used.
+        furl_or_file : str
+            A furl or a filename containing a FURLK. This is useful if you 
+            simply know the location of the FURL file.
+        ipythondir : str
+            The location of the ipythondir if different from the default.
+            This is used if the cluster directory is being found by profile.
                 
-        :Returns:
-            A deferred to the actual client class
+        Returns
+        -------
+        A deferred to the actual client class.
         """
-        task_co = client_co['client_interfaces']['multiengine']
-        if furl_or_file:
-            ff = furl_or_file
-        else:
-            ff = task_co['furl_file']
-        return self.get_client(ff)
+        return self.get_client(
+            profile, cluster_dir, furl_or_file, 
+            'ipcontroller-mec.furl', ipythondir
+        )
     
-    def get_client(self, furl_or_file):
-        """
-        Get a remote reference and wrap it in a client by furl.
-        
-        This method first gets a remote reference and then calls its 
-        `get_client_name` method to find the apprpriate client class
-        that should be used to wrap the remote reference.
-        
-        :Parameters:
-            furl_or_file : str
-                A furl or a filename containing a furl
+    def get_client(self, profile='default', cluster_dir=None,
+                   furl_or_file=None, furl_file_name=None, ipythondir=None):
+        """Get a remote reference and wrap it in a client by furl.
+
+        This method is a simple wrapper around `get_client` that passes in
+        the default name of the task client FURL file.  Usually only
+        the ``profile`` option will be needed.  If a FURL file can't be
+        found by its profile, use ``cluster_dir`` or ``furl_or_file``.
         
-        :Returns:
-            A deferred to the actual client class
+        Parameters
+        ----------
+        profile : str
+            The name of a cluster directory profile (default="default"). The
+            cluster directory "cluster_<profile>" will be searched for
+            in ``os.getcwd()``, the ipythondir and then in the directories
+            listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
+        cluster_dir : str
+            The full path to a cluster directory.  This is useful if profiles
+            are not being used.
+        furl_or_file : str
+            A furl or a filename containing a FURLK. This is useful if you 
+            simply know the location of the FURL file.
+        furl_file_name : str
+            The filename (not the full path) of the FURL. This must be
+            provided if ``furl_or_file`` is not.
+        ipythondir : str
+            The location of the ipythondir if different from the default.
+            This is used if the cluster directory is being found by profile.
+
+        Returns
+        -------
+        A deferred to the actual client class.
         """
-        furl = find_furl(furl_or_file)
+        try:
+            furl = self._find_furl(
+                profile, cluster_dir, furl_or_file,
+                furl_file_name, ipythondir
+            )
+        except:
+            return defer.fail(failure.Failure())
+
         d = self.get_reference(furl)
-        def wrap_remote_reference(rr):
+
+        def _wrap_remote_reference(rr):
             d = rr.callRemote('get_client_name')
             d.addCallback(lambda name: import_item(name))
             def adapt(client_interface):
@@ -146,5 +233,352 b' class ClientConnector(object):'
             d.addCallback(adapt)
 
             return d
-        d.addCallback(wrap_remote_reference)
+
+        d.addCallback(_wrap_remote_reference)
         return d
+
+
+class ClientConnector(object):
+    """A blocking version of a client connector.
+
+    This class creates a single :class:`Tub` instance and allows remote
+    references and client to be retrieved by their FURLs.  Remote references
+    are cached locally and FURL files can be found using profiles and cluster
+    directories.
+    """
+
+    def __init__(self):
+        self.async_cc = AsyncClientConnector()
+
+    def get_task_client(self, profile='default', cluster_dir=None,
+                        furl_or_file=None, ipythondir=None):
+        """Get the task client.
+        
+        Usually only the ``profile`` option will be needed. If a FURL file
+        can't be found by its profile, use ``cluster_dir`` or
+        ``furl_or_file``.
+        
+        Parameters
+        ----------
+        profile : str
+            The name of a cluster directory profile (default="default"). The
+            cluster directory "cluster_<profile>" will be searched for
+            in ``os.getcwd()``, the ipythondir and then in the directories
+            listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
+        cluster_dir : str
+            The full path to a cluster directory.  This is useful if profiles
+            are not being used.
+        furl_or_file : str
+            A furl or a filename containing a FURLK. This is useful if you 
+            simply know the location of the FURL file.
+        ipythondir : str
+            The location of the ipythondir if different from the default.
+            This is used if the cluster directory is being found by profile.
+
+        Returns
+        -------
+        The task client instance.
+        """
+        client = blockingCallFromThread(
+            self.async_cc.get_task_client, profile, cluster_dir,
+            furl_or_file, ipythondir
+        )
+        return client.adapt_to_blocking_client()
+
+    def get_multiengine_client(self, profile='default', cluster_dir=None,
+                               furl_or_file=None, ipythondir=None):
+        """Get the multiengine client.
+        
+        Usually only the ``profile`` option will be needed. If a FURL file
+        can't be found by its profile, use ``cluster_dir`` or
+        ``furl_or_file``.
+        
+        Parameters
+        ----------
+        profile : str
+            The name of a cluster directory profile (default="default"). The
+            cluster directory "cluster_<profile>" will be searched for
+            in ``os.getcwd()``, the ipythondir and then in the directories
+            listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
+        cluster_dir : str
+            The full path to a cluster directory.  This is useful if profiles
+            are not being used.
+        furl_or_file : str
+            A furl or a filename containing a FURLK. This is useful if you 
+            simply know the location of the FURL file.
+        ipythondir : str
+            The location of the ipythondir if different from the default.
+            This is used if the cluster directory is being found by profile.
+
+        Returns
+        -------
+        The multiengine client instance.
+        """
+        client = blockingCallFromThread(
+            self.async_cc.get_multiengine_client, profile, cluster_dir,
+            furl_or_file, ipythondir
+        )
+        return client.adapt_to_blocking_client()
+
+    def get_client(self, profile='default', cluster_dir=None,
+                   furl_or_file=None, ipythondir=None):
+        client = blockingCallFromThread(
+            self.async_cc.get_client, profile, cluster_dir,
+            furl_or_file, ipythondir
+        )
+        return client.adapt_to_blocking_client()
+
+
+class ClusterStateError(Exception):
+    pass
+
+
+class AsyncCluster(object):
+    """An class that wraps the :command:`ipcluster` script."""
+
+    def __init__(self, profile='default', cluster_dir=None, ipythondir=None,
+                 auto_create=False, auto_stop=True):
+        """Create a class to manage an IPython cluster.
+
+        This class calls the :command:`ipcluster` command with the right
+        options to start an IPython cluster.  Typically a cluster directory
+        must be created (:command:`ipcluster create`) and configured before 
+        using this class. Configuration is done by editing the 
+        configuration files in the top level of the cluster directory.
+
+        Parameters
+        ----------
+        profile : str
+            The name of a cluster directory profile (default="default"). The
+            cluster directory "cluster_<profile>" will be searched for
+            in ``os.getcwd()``, the ipythondir and then in the directories
+            listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
+        cluster_dir : str
+            The full path to a cluster directory.  This is useful if profiles
+            are not being used.
+        furl_or_file : str
+            A furl or a filename containing a FURLK. This is useful if you 
+            simply know the location of the FURL file.
+        ipythondir : str
+            The location of the ipythondir if different from the default.
+            This is used if the cluster directory is being found by profile.
+        auto_create : bool
+            Automatically create the cluster directory it is dones't exist.
+            This will usually only make sense if using a local cluster 
+            (default=False).
+        auto_stop : bool
+            Automatically stop the cluster when this instance is garbage 
+            collected (default=True).  This is useful if you want the cluster
+            to live beyond your current process. There is also an instance
+            attribute ``auto_stop`` to change this behavior.
+        """
+        self._setup_cluster_dir(profile, cluster_dir, ipythondir, auto_create)
+        self.state = 'before'
+        self.launcher = None
+        self.client_connector = None
+        self.auto_stop = auto_stop
+
+    def __del__(self):
+        if self.auto_stop and self.state=='running':
+            print "Auto stopping the cluster..."
+            self.stop()
+
+    @property
+    def location(self):
+        if hasattr(self, 'cluster_dir_obj'):
+            return self.cluster_dir_obj.location
+        else:
+            return ''
+
+    @property
+    def running(self):
+        if self.state=='running':
+            return True
+        else:
+            return False
+
+    def _setup_cluster_dir(self, profile, cluster_dir, ipythondir, auto_create):
+        if ipythondir is None:
+            ipythondir = get_ipython_dir()
+        if cluster_dir is not None:
+            try:
+                self.cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
+            except ClusterDirError:
+                pass
+        if profile is not None:
+            try:
+                self.cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
+                    ipythondir, profile)
+            except ClusterDirError:
+                pass
+        if auto_create or profile=='default':
+            # This should call 'ipcluster create --profile default
+            self.cluster_dir_obj = ClusterDir.create_cluster_dir_by_profile(
+                ipythondir, profile)
+        else:
+            raise ClusterDirError('Cluster dir not found.')
+
+    @make_deferred
+    def start(self, n=2):
+        """Start the IPython cluster with n engines.
+
+        Parameters
+        ----------
+        n : int
+            The number of engine to start.
+        """
+        # We might want to add logic to test if the cluster has started
+        # by another process....
+        if not self.state=='running':
+            self.launcher = IPClusterLauncher(os.getcwd())
+            self.launcher.ipcluster_n = n
+            self.launcher.ipcluster_subcommand = 'start'
+            d = self.launcher.start()
+            d.addCallback(self._handle_start)
+            return d
+        else:
+            raise ClusterStateError('Cluster is already running')
+
+    @make_deferred
+    def stop(self):
+        """Stop the IPython cluster if it is running."""
+        if self.state=='running':
+            d1 = self.launcher.observe_stop()
+            d1.addCallback(self._handle_stop)
+            d2 = self.launcher.stop()
+            return gatherBoth([d1, d2], consumeErrors=True)
+        else:
+            raise ClusterStateError("Cluster not running")
+
+    def get_multiengine_client(self):
+        """Get the multiengine client for the running cluster.
+
+        If this fails, it means that the cluster has not finished starting.
+        Usually waiting a few seconds are re-trying will solve this.    
+        """
+        if self.client_connector is None:
+            self.client_connector = AsyncClientConnector()
+        return self.client_connector.get_multiengine_client(
+            cluster_dir=self.cluster_dir_obj.location
+        )
+
+    def get_task_client(self):
+        """Get the task client for the running cluster.
+
+        If this fails, it means that the cluster has not finished starting.
+        Usually waiting a few seconds are re-trying will solve this.    
+        """
+        if self.client_connector is None:
+            self.client_connector = AsyncClientConnector()
+        return self.client_connector.get_task_client(
+            cluster_dir=self.cluster_dir_obj.location
+        )
+
+    def _handle_start(self, r):
+        self.state = 'running'
+
+    def _handle_stop(self, r):
+        self.state = 'after'
+
+
+class Cluster(object):
+
+
+    def __init__(self, profile='default', cluster_dir=None, ipythondir=None,
+                 auto_create=False, auto_stop=True):
+        """Create a class to manage an IPython cluster.
+
+        This class calls the :command:`ipcluster` command with the right
+        options to start an IPython cluster.  Typically a cluster directory
+        must be created (:command:`ipcluster create`) and configured before 
+        using this class. Configuration is done by editing the 
+        configuration files in the top level of the cluster directory.
+
+        Parameters
+        ----------
+        profile : str
+            The name of a cluster directory profile (default="default"). The
+            cluster directory "cluster_<profile>" will be searched for
+            in ``os.getcwd()``, the ipythondir and then in the directories
+            listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
+        cluster_dir : str
+            The full path to a cluster directory.  This is useful if profiles
+            are not being used.
+        furl_or_file : str
+            A furl or a filename containing a FURLK. This is useful if you 
+            simply know the location of the FURL file.
+        ipythondir : str
+            The location of the ipythondir if different from the default.
+            This is used if the cluster directory is being found by profile.
+        auto_create : bool
+            Automatically create the cluster directory it is dones't exist.
+            This will usually only make sense if using a local cluster 
+            (default=False).
+        auto_stop : bool
+            Automatically stop the cluster when this instance is garbage 
+            collected (default=True).  This is useful if you want the cluster
+            to live beyond your current process. There is also an instance
+            attribute ``auto_stop`` to change this behavior.
+        """
+        self.async_cluster = AsyncCluster(
+            profile, cluster_dir, ipythondir, auto_create, auto_stop
+        )
+        self.cluster_dir_obj = self.async_cluster.cluster_dir_obj
+        self.client_connector = None
+
+    def _set_auto_stop(self, value):
+        self.async_cluster.auto_stop = value
+
+    def _get_auto_stop(self):
+        return self.async_cluster.auto_stop
+
+    auto_stop = property(_get_auto_stop, _set_auto_stop)
+
+    @property
+    def location(self):
+        return self.async_cluster.location
+
+    @property
+    def running(self):
+        return self.async_cluster.running
+
+    def start(self, n=2):
+        """Start the IPython cluster with n engines.
+
+        Parameters
+        ----------
+        n : int
+            The number of engine to start.
+        """
+        return blockingCallFromThread(self.async_cluster.start, n)
+
+    def stop(self):
+        """Stop the IPython cluster if it is running."""
+        return blockingCallFromThread(self.async_cluster.stop)
+
+    def get_multiengine_client(self):
+        """Get the multiengine client for the running cluster.
+
+        If this fails, it means that the cluster has not finished starting.
+        Usually waiting a few seconds are re-trying will solve this.    
+        """
+        if self.client_connector is None:
+            self.client_connector = ClientConnector()
+        return self.client_connector.get_multiengine_client(
+            cluster_dir=self.cluster_dir_obj.location
+        )
+
+    def get_task_client(self):
+        """Get the task client for the running cluster.
+
+        If this fails, it means that the cluster has not finished starting.
+        Usually waiting a few seconds are re-trying will solve this.    
+        """
+        if self.client_connector is None:
+            self.client_connector = ClientConnector()
+        return self.client_connector.get_task_client(
+            cluster_dir=self.cluster_dir_obj.location
+        )
+
+
+

IPython/kernel/clusterdir.py

@@ -17,6 +17,9 b' The IPython cluster directory'
 
 import os
 import shutil
+import sys
+
+from twisted.python import log
 
 from IPython.core import release
 from IPython.config.loader import PyFileConfigLoader
@@ -210,7 +213,8 b' class AppWithClusterDirArgParseConfigLoader(ArgParseConfigLoader):'
             dest='Global.ipythondir',type=str,
             help='Set to override default location of Global.ipythondir.',
             default=NoConfigDefault,
-            metavar='Global.ipythondir')
+            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 '
@@ -218,19 +222,31 b' class AppWithClusterDirArgParseConfigLoader(ArgParseConfigLoader):'
             'is named "default".  The cluster directory is resolve this way '
             'if the --cluster-dir option is not used.',
             default=NoConfigDefault,
-            metavar='Global.profile')
+            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,
-            metavar="Global.log_level")
+            metavar="Global.log_level"
+        )
         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')
-
+            metavar='Global.cluster_dir'
+        )
+        self.parser.add_argument('-clean_logs', '--clean-logs',
+            dest='Global.clean_logs', action='store_true',
+            help='Delete old log flies before starting.',
+            default=NoConfigDefault
+        )
+        self.parser.add_argument('-noclean_logs', '--no-clean-logs',
+            dest='Global.clean_logs', action='store_false',
+            help="Don't Delete old log flies before starting.",
+            default=NoConfigDefault
+        )
 
 class ApplicationWithClusterDir(Application):
     """An application that puts everything into a cluster directory.
@@ -257,6 +273,8 b' class ApplicationWithClusterDir(Application):'
         super(ApplicationWithClusterDir, self).create_default_config()
         self.default_config.Global.profile = 'default'
         self.default_config.Global.cluster_dir = ''
+        self.default_config.Global.log_to_file = False
+        self.default_config.Global.clean_logs = False
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
@@ -349,4 +367,28 b' class ApplicationWithClusterDir(Application):'
         # Set the search path to the cluster directory
         self.config_file_paths = (self.cluster_dir,)
 
-
+    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)
+
+    def start_logging(self):
+        # Remove old log files
+        if self.master_config.Global.clean_logs:
+            log_dir = self.master_config.Global.log_dir
+            for f in os.listdir(log_dir):
+                if f.startswith(self.name + '-') and f.endswith('.log'):
+                    os.remove(os.path.join(log_dir, f))
+        # Start logging to the new log file
+        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)

IPython/kernel/engineconnector.py

@@ -1,19 +1,18 b''
+#!/usr/bin/env python
 # encoding: utf-8
 
 """A class that manages the engines connection to the controller."""
 
-__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
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 
 import os
 import cPickle as pickle
@@ -26,9 +25,9 b' 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):

IPython/kernel/ipclusterapp.py

@@ -122,7 +122,16 b' class IPClusterCLLoader(ArgParseConfigLoader):'
             help='The number of engines to start.',
             metavar='Global.n'
         )
-
+        parser_start.add_argument('-clean_logs', '--clean-logs',
+            dest='Global.clean_logs', action='store_true',
+            help='Delete old log flies before starting.',
+            default=NoConfigDefault
+        )
+        parser_start.add_argument('-noclean_logs', '--no-clean-logs',
+            dest='Global.clean_logs', action='store_false',
+            help="Don't delete old log flies before starting.",
+            default=NoConfigDefault
+        )
 
 default_config_file_name = 'ipcluster_config.py'
 
@@ -141,9 +150,9 b' class IPClusterApp(ApplicationWithClusterDir):'
             'IPython.kernel.launcher.LocalControllerLauncher'
         self.default_config.Global.engine_launcher = \
             'IPython.kernel.launcher.LocalEngineSetLauncher'
-        self.default_config.Global.log_to_file = False
         self.default_config.Global.n = 2
         self.default_config.Global.reset_config = False
+        self.default_config.Global.clean_logs = True
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
@@ -172,6 +181,7 b' class IPClusterApp(ApplicationWithClusterDir):'
                     "'ipcluster create -h' or 'ipcluster list -h' for more "
                     "information about creating and listing cluster dirs."
                 )
+
     def construct(self):
         config = self.master_config
         if config.Global.subcommand=='list':
@@ -184,15 +194,21 b' class IPClusterApp(ApplicationWithClusterDir):'
             self.start_logging()
             reactor.callWhenRunning(self.start_launchers)
 
-    def list_cluster_dirs(self):        
+    def list_cluster_dirs(self):
+        # Find the search paths
         cluster_dir_paths = os.environ.get('IPCLUSTERDIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
-        # We need to look both in default_config and command_line_config!!!
-        paths = [os.getcwd(), self.default_config.Global.ipythondir] + \
+        try:
+            ipythondir = self.command_line_config.Global.ipythondir
+        except AttributeError:
+            ipythondir = self.default_config.Global.ipythondir
+        paths = [os.getcwd(), ipythondir] + \
             cluster_dir_paths
+        paths = list(set(paths))
+
         self.log.info('Searching for cluster dirs in paths: %r' % paths)
         for path in paths:
             files = os.listdir(path)
@@ -203,15 +219,6 b' class IPClusterApp(ApplicationWithClusterDir):'
                     start_cmd = '"ipcluster start -n 4 -p %s"' % profile
                     print start_cmd + " ==> " + full_path
 
-    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 start_launchers(self):
         config = self.master_config
 
@@ -227,6 +234,7 b' class IPClusterApp(ApplicationWithClusterDir):'
 
         # Setup signals
         signal.signal(signal.SIGINT, self.stop_launchers)
+        # signal.signal(signal.SIGKILL, self.stop_launchers)
 
         # Setup the observing of stopping
         d1 = self.controller_launcher.observe_stop()
@@ -261,10 +269,24 b' class IPClusterApp(ApplicationWithClusterDir):'
     def stop_launchers(self, signum, frame):
         log.msg("Stopping cluster")
         d1 = self.engine_launcher.stop()
-        d1.addCallback(lambda _: self.controller_launcher.stop)
+        d2 = self.controller_launcher.stop()
+        # d1.addCallback(lambda _: self.controller_launcher.stop)
         d1.addErrback(self.err_and_stop)
+        d2.addErrback(self.err_and_stop)
         reactor.callLater(2.0, reactor.stop)
 
+    def start_logging(self):
+        # Remove old log files
+        if self.master_config.Global.clean_logs:
+            log_dir = self.master_config.Global.log_dir
+            for f in os.listdir(log_dir):
+                if f.startswith('ipengine' + '-') and f.endswith('.log'):
+                    os.remove(os.path.join(log_dir, f))
+            for f in os.listdir(log_dir):
+                if f.startswith('ipcontroller' + '-') and f.endswith('.log'):
+                    os.remove(os.path.join(log_dir, f))
+        super(IPClusterApp, self).start_logging()
+
     def start_app(self):
         config = self.master_config
         if config.Global.subcommand=='create' or config.Global.subcommand=='list':
@@ -280,4 +302,5 b' def launch_new_instance():'
 
 
 if __name__ == '__main__':
-    launch_new_instance()
  No newline at end of file
+    launch_new_instance()
+

IPython/kernel/ipcontrollerapp.py

@@ -185,7 +185,7 b' class IPControllerApp(ApplicationWithClusterDir):'
         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
+        self.default_config.Global.clean_logs = True
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
@@ -206,18 +206,6 b' class IPControllerApp(ApplicationWithClusterDir):'
             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
@@ -240,15 +228,6 b' class IPControllerApp(ApplicationWithClusterDir):'
         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:

IPython/kernel/ipengineapp.py

@@ -97,12 +97,12 b' class IPEngineApp(ApplicationWithClusterDir):'
     def create_default_config(self):
         super(IPEngineApp, self).create_default_config()
 
+        # The engine should not clean logs as we don't want to remove the
+        # active log files of other running engines.
+        self.default_config.Global.clean_logs = False
+
         # 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
@@ -113,6 +113,11 b' class IPEngineApp(ApplicationWithClusterDir):'
         # If not, this is computed using the profile, app_dir and furl_file_name
         self.default_config.Global.furl_file = ''
 
+        # The max number of connection attemps and the initial delay between
+        # those attemps.
+        self.default_config.Global.connect_delay = 0.1
+        self.default_config.Global.connect_max_tries = 15
+
         # MPI related config attributes
         self.default_config.MPI.use = ''
         self.default_config.MPI.mpi4py = mpi4py_init
@@ -129,15 +134,7 b' class IPEngineApp(ApplicationWithClusterDir):'
         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)
-
+        super(IPEngineApp, self).pre_construct()
         self.find_cont_furl_file()
 
     def find_cont_furl_file(self):
@@ -189,7 +186,9 b' class IPEngineApp(ApplicationWithClusterDir):'
     def call_connect(self):
         d = self.engine_connector.connect_to_controller(
             self.engine_service, 
-            self.master_config.Global.furl_file
+            self.master_config.Global.furl_file,
+            self.master_config.Global.connect_delay,
+            self.master_config.Global.connect_max_tries
         )
 
         def handle_error(f):
@@ -216,15 +215,6 b' class IPEngineApp(ApplicationWithClusterDir):'
         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:

IPython/kernel/launcher.py

@@ -475,7 +475,9 b' def find_engine_cmd():'
 class LocalEngineLauncher(LocalProcessLauncher):
 
     engine_cmd = List(find_engine_cmd())
-    engine_args = List(['--log-to-file','--log-level', '40'], config=True)
+    engine_args = List(
+        ['--log-to-file','--log-level', '40'], config=True
+    )
 
     def find_args(self):
         return self.engine_cmd + self.engine_args
@@ -490,7 +492,9 b' class LocalEngineLauncher(LocalProcessLauncher):'
 
 class LocalEngineSetLauncher(BaseLauncher):
 
-    engine_args = List(['--log-to-file','--log-level', '40'], config=True)
+    engine_args = List(
+        ['--log-to-file','--log-level', '40'], config=True
+    )
 
     def __init__(self, working_dir, parent=None, name=None, config=None):
         super(LocalEngineSetLauncher, self).__init__(
@@ -547,7 +551,9 b' class LocalEngineSetLauncher(BaseLauncher):'
 class MPIExecEngineSetLauncher(MPIExecLauncher):
 
     engine_cmd = List(find_engine_cmd(), config=False)
-    engine_args = List(['--log-to-file','--log-level', '40'], config=True)
+    engine_args = List(
+        ['--log-to-file','--log-level', '40'], config=True
+    )    
     n = Int(1, config=True)
 
     def start(self, n, profile=None, cluster_dir=None):
@@ -582,4 +588,41 b' class SSHEngineSetLauncher(BaseLauncher):'
     pass
 
 
+#-----------------------------------------------------------------------------
+# A launcher for ipcluster itself!
+#-----------------------------------------------------------------------------
+
+
+def find_ipcluster_cmd():
+    if sys.platform == 'win32':
+        # This logic is needed because the ipcluster script doesn't
+        # always get installed in the same way or in the same location.
+        from IPython.kernel import ipclusterapp
+        script_location = ipclusterapp.__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.
+        cmd = [sys.executable, '-u', script_location]
+    else:
+        # ipcontroller has to be on the PATH in this case.
+        cmd = ['ipcluster']
+    return cmd
+
+
+class IPClusterLauncher(LocalProcessLauncher):
+
+    ipcluster_cmd = List(find_ipcluster_cmd())
+    ipcluster_args = List(
+        ['--clean-logs', '--log-to-file', '--log-level', '40'], config=True)
+    ipcluster_subcommand = Str('start')
+    ipcluster_n = Int(2)
+
+    def find_args(self):
+        return self.ipcluster_cmd + [self.ipcluster_subcommand] + \
+            ['-n', repr(self.ipcluster_n)] + self.ipcluster_args
+
+    def start(self):
+        log.msg("Starting ipcluster: %r" % self.args)
+        return super(IPClusterLauncher, self).start()
 

IPython/kernel/twistedutil.py

@@ -3,18 +3,16 b''
 
 """Things directly related to all of twisted."""
 
-__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
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 
 import os, sys
 import threading, Queue, atexit
@@ -25,9 +23,9 b' from twisted.python import log, failure'
 
 from IPython.kernel.error import FileTimeoutError
 
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 # Classes related to twisted and threads
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
 
 
 class ReactorInThread(threading.Thread):