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

@@ -1,66 +1,67 b''
 import os
 
 c = get_config()
 
 # Options are:
 # * LocalControllerLauncher
 # * PBSControllerLauncher
 # c.Global.controller_launcher = 'IPython.kernel.launcher.LocalControllerLauncher'
 
 # Options are:
 # * LocalEngineSetLauncher
 # * MPIExecEngineSetLauncher
 # * PBSEngineSetLauncher
 # c.Global.engine_launcher = 'IPython.kernel.launcher.LocalEngineSetLauncher'
 
 # 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 = []
 # c.MPIExecLauncher.program = []
 # c.MPIExecLauncher.program_args = []
 # c.MPIExecLauncher.n = 1
 
 # c.SSHLauncher.ssh_cmd = ['ssh']
 # c.SSHLauncher.ssh_args = []
 # c.SSHLauncher.program = []
 # s.SSHLauncher.program_args = []
 # c.SSHLauncher.hostname = ''
 # c.SSHLauncher.user = os.environ['USER']
 
 # c.PBSLauncher.submit_command = 'qsub'
 # c.PBSLauncher.delete_command = 'qdel'
 # c.PBSLauncher.job_id_regexp = '\d+'
 # c.PBSLauncher.batch_template = """"""
 # c.PBSLauncher.batch_file_name = u'pbs_batch_script'
 
 # c.LocalControllerLauncher.controller_args = []
 
 # c.MPIExecControllerLauncher.mpi_cmd = ['mpiexec']
 # c.MPIExecControllerLauncher.mpi_args = []
 # c.MPIExecControllerLauncher.controller_args = []
 # c.MPIExecControllerLauncher.n = 1
 
 # c.PBSControllerLauncher.submit_command = 'qsub'
 # c.PBSControllerLauncher.delete_command = 'qdel'
 # c.PBSControllerLauncher.job_id_regexp = '\d+'
 # c.PBSControllerLauncher.batch_template = """"""
 # c.PBSLauncher.batch_file_name = u'pbs_batch_script'
 
 # c.LocalEngineLauncher.engine_args = []
 
 # c.LocalEngineSetLauncher.engine_args = []
 
 # c.MPIExecEngineSetLauncher.mpi_cmd = ['mpiexec']
 # c.MPIExecEngineSetLauncher.mpi_args = []
 # c.MPIExecEngineSetLauncher.controller_args = []
 # c.MPIExecEngineSetLauncher.n = 1
 
 # c.PBSEngineSetLauncher.submit_command = 'qsub'
 # c.PBSEngineSetLauncher.delete_command = 'qdel'
 # c.PBSEngineSetLauncher.job_id_regexp = '\d+'
 # c.PBSEngineSetLauncher.batch_template = """"""
 # c.PBSEngineSetLauncher.batch_file_name = u'pbs_batch_script'
 

IPython/config/default/ipcontroller_config.py

@@ -1,72 +1,68 b''
 from IPython.config.loader import Config
 
 c = get_config()
 
 #-----------------------------------------------------------------------------
 # Global configuration
 #-----------------------------------------------------------------------------
 
 # 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
 #-----------------------------------------------------------------------------
 
 # Basic client service config attributes
 # c.FCClientServiceFactory.ip = ''
 # c.FCClientServiceFactory.port = 0
 # c.FCClientServiceFactory.location = ''
 # c.FCClientServiceFactory.secure =  True
 # c.FCClientServiceFactory.reuse_furls = False
 
 # You shouldn't have to modify the rest of this section
 # c.FCClientServiceFactory.cert_file = 'ipcontroller-client.pem'
 
 # 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'
 # 
 # c.FCEngineServiceFactory.interfaces = default_client_interfaces
 
 #-----------------------------------------------------------------------------
 # Configure the engine services
 #-----------------------------------------------------------------------------
 
 # Basic config attributes for the engine services
 # c.FCEngineServiceFactory.ip = ''
 # c.FCEngineServiceFactory.port = 0
 # c.FCEngineServiceFactory.location = ''
 # c.FCEngineServiceFactory.secure = True
 # c.FCEngineServiceFactory.reuse_furls = False
 
 # You shouldn't have to modify the rest of this section
 # c.FCEngineServiceFactory.cert_file = 'ipcontroller-engine.pem'
 
 # default_engine_interfaces = Config()
 # default_engine_interfaces.Default.interface_chain = [
 #     'IPython.kernel.enginefc.IFCControllerBase'
 # ]
 # 
 # default_engine_interfaces.Default.furl_file = 'ipcontroller-engine.furl'
 # 
 # c.FCEngineServiceFactory.interfaces = default_engine_interfaces

IPython/config/default/ipengine_config.py

@@ -1,24 +1,28 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
 # mpi.size = mpi.COMM_WORLD.Get_size()
 # mpi.rank = mpi.COMM_WORLD.Get_rank()
 # """
 # c.MPI.pytrilinos = """from PyTrilinos import Epetra
 # class SimpleStruct:
 # pass
 # mpi = SimpleStruct()
 # mpi.rank = 0
 # mpi.size = 0
 # """
 

IPython/kernel/asyncclient.py

@@ -1,41 +1,42 b''
+#!/usr/bin/env python
 # encoding: utf-8
 
 """Asynchronous clients for the IPython controller.
 
 This module has clients for using the various interfaces of the controller 
 in a fully asynchronous manner.  This means that you will need to run the
 Twisted reactor yourself and that all methods of the client classes return
 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,96 +1,87 b''
+#!/usr/bin/env python
 # encoding: utf-8
 
 """This module contains blocking clients for the controller interfaces.
 
 Unlike the clients in `asyncclient.py`, the clients in this module are fully
 blocking.  This means that methods on the clients return the actual results
 rather than a deferred to the result.  Also, we manage the Twisted reactor
 for you.  This is done by running the reactor in a thread.
 
 The main classes in this module are:
 
     * MultiEngineClient
     * TaskClient
     * Task
     * 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
 
 # These enable various things 
 from IPython.kernel import codeutil
 # import IPython.kernel.magic
 
 # Other things that the user will need
 from IPython.kernel.task import MapTask, StringTask
 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()
+
+
+
+
+__all__ = [
+    'MapTask',
+    'StringTask',
+    'MultiEngineClient',
+    'TaskClient',
+    'CompositeError',
+    'get_task_client',
+    'get_multiengine_client',
+    'Cluster'
+]

IPython/kernel/clientconnector.py

@@ -1,150 +1,584 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:
+        Parameters
+        ----------
             furl_or_file : str
                 A furl or a filename containing a furl
         
-        :Returns:
+        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:
+        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 furl.  If empty, the
-                default furl_file will be used
+            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']['task']
-        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-tc.furl', ipythondir
+        )
 
-    def get_multiengine_client(self, furl_or_file=''):
-        """
-        Get the multiengine controller client.
+    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:
+        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 furl.  If empty, the
-                default furl_file will be used
+            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.
+    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 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.
+        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:
+        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 furl
+            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
+        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):
                 client = client_interface(rr)
                 client.tub = self.tub
                 return client
             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

@@ -1,352 +1,394 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
+import sys
+
+from twisted.python import log
 
 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, Bool
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 
 class ClusterDirError(Exception):
     pass
 
 
 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):
         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(self.log_dir, 0777)
 
     def _security_dir_changed(self, name, old, new):
         self.check_security_dir()
 
     def check_security_dir(self):
         if not os.path.isdir(self.security_dir):
             os.mkdir(self.security_dir, 0700)
         else:
             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, overwrite=False):
         """Copy a default config file into the active cluster directory.
 
         Default configuration files are kept in :mod:`IPython.config.default`.
         This function moves these from that location to the working cluster
         directory.
         """
         if path is None:
             import IPython.config.default
             path = IPython.config.default.__file__.split(os.path.sep)[:-1]
             path = os.path.sep.join(path)
         src = os.path.join(path, config_file)
         dst = os.path.join(self.location, config_file)
         if not os.path.isfile(dst) or overwrite:
             shutil.copy(src, dst)
 
     def copy_all_config_files(self, path=None, overwrite=False):
         """Copy all config files into the active cluster directory."""
         for f in ['ipcontroller_config.py', 'ipengine_config.py',
                   'ipcluster_config.py']:
             self.copy_config_file(f, path=path, overwrite=overwrite)
 
     @classmethod
     def create_cluster_dir(csl, cluster_dir):
         """Create a new cluster directory given a full path.
 
         Parameters
         ----------
         cluster_dir : str
             The full path to the cluster directory.  If it does exist, it will
             be used.  If not, it will be created.
         """
         return ClusterDir(cluster_dir)
 
     @classmethod
     def create_cluster_dir_by_profile(cls, path, profile='default'):
         """Create a cluster dir by profile name and path.
 
         Parameters
         ----------
         path : str
             The path (directory) to put the cluster directory in.
         profile : str
             The name of the profile.  The name of the cluster directory will
             be "cluster_<profile>".
         """
         if not os.path.isdir(path):
             raise ClusterDirError('Directory not found: %s' % path)
         cluster_dir = os.path.join(path, 'cluster_' + profile)
         return ClusterDir(cluster_dir)
 
     @classmethod
     def find_cluster_dir_by_profile(cls, ipythondir, profile='default'):
         """Find an existing cluster dir by profile name, return its ClusterDir.
 
         This searches through a sequence of paths for a cluster dir.  If it
         is not found, a :class:`ClusterDirError` exception will be raised.
 
         The search path algorithm is:
         1. ``os.getcwd()``
         2. ``ipythondir``
         3. The directories found in the ":" separated 
            :env:`IPCLUSTERDIR_PATH` environment variable.
 
         Parameters
         ----------
         ipythondir : unicode or str
             The IPython directory to use.
         profile : unicode or str
             The name of the profile.  The name of the cluster directory
             will be "cluster_<profile>".
         """
         dirname = 'cluster_' + profile
         cluster_dir_paths = os.environ.get('IPCLUSTERDIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
         paths = [os.getcwd(), ipythondir] + cluster_dir_paths
         for p in paths:
             cluster_dir = os.path.join(p, dirname)
             if os.path.isdir(cluster_dir):
                 return ClusterDir(cluster_dir)
         else:
             raise ClusterDirError('Cluster directory not found in paths: %s' % dirname)
 
     @classmethod
     def find_cluster_dir(cls, cluster_dir):
         """Find/create a cluster dir and return its ClusterDir.
 
         This will create the cluster directory if it doesn't exist.
 
         Parameters
         ----------
         cluster_dir : unicode or str
             The path of the cluster directory.  This is expanded using
             :func:`os.path.expandvars` and :func:`os.path.expanduser`.
         """
         cluster_dir = os.path.expandvars(os.path.expanduser(cluster_dir))
         if not os.path.isdir(cluster_dir):
             raise ClusterDirError('Cluster directory not found: %s' % 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')
+            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')
+            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.
 
     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.
     """
 
     auto_create_cluster_dir = True
 
     def create_default_config(self):
         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."""
         return AppWithClusterDirArgParseConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def find_resources(self):
         """This resolves the cluster directory.
 
         This tries to find the cluster directory and if successful, it will
         have done:
         * Sets ``self.cluster_dir_obj`` to the :class:`ClusterDir` object for 
           the application.
         * Sets ``self.cluster_dir`` attribute of the application and config
           objects.
 
         The algorithm used for this is as follows:
         1. Try ``Global.cluster_dir``.
         2. Try using ``Global.profile``.
         3. If both of these fail and ``self.auto_create_cluster_dir`` is
            ``True``, then create the new cluster dir in the IPython directory.
         4. If all fails, then raise :class:`ClusterDirError`.
         """
 
         try:
             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))
         try:
             self.cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
         except ClusterDirError:
             pass
         else:
             self.log.info('Using existing cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
             return
 
         try:
             self.profile = self.command_line_config.Global.profile
         except AttributeError:
             self.profile = self.default_config.Global.profile
         try:
             self.cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
                 self.ipythondir, self.profile)
         except ClusterDirError:
             pass
         else:
             self.log.info('Using existing cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
             return
 
         if self.auto_create_cluster_dir:
             self.cluster_dir_obj = ClusterDir.create_cluster_dir_by_profile(
                 self.ipythondir, self.profile
             )
             self.log.info('Creating new cluster dir: %s' % \
                 self.cluster_dir_obj.location
             )
             self.finish_cluster_dir()
         else:
             raise ClusterDirError('Could not find a valid cluster directory.')
 
     def finish_cluster_dir(self):
         # 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
 
         # Set the search path to the cluster directory
         self.config_file_paths = (self.cluster_dir,)
 
     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):
         # 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,131 +1,130 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
 
 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,
                               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.
 
         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)
 
         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:
             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.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/ipclusterapp.py

@@ -1,283 +1,306 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The ipcluster application.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import logging
 import os
 import signal
 import sys
 
 from IPython.core import release
 from IPython.external import argparse
 from IPython.config.loader import ArgParseConfigLoader, NoConfigDefault
 from IPython.utils.importstring import import_item
 
 from IPython.kernel.clusterdir import (
     ApplicationWithClusterDir, ClusterDirError
 )
 
 from twisted.internet import reactor, defer
 from twisted.python import log
 
 #-----------------------------------------------------------------------------
 # Code for launchers
 #-----------------------------------------------------------------------------
 
 
 
 #-----------------------------------------------------------------------------
 # The ipcluster application
 #-----------------------------------------------------------------------------
 
 
 class IPClusterCLLoader(ArgParseConfigLoader):
 
     def _add_arguments(self):
         # This has all the common options that all subcommands use
         parent_parser1 = argparse.ArgumentParser(add_help=False)
         parent_parser1.add_argument('-ipythondir', '--ipython-dir', 
             dest='Global.ipythondir',type=str,
             help='Set to override default location of Global.ipythondir.',
             default=NoConfigDefault,
             metavar='Global.ipythondir')
         parent_parser1.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')
 
         # This has all the common options that other subcommands use
         parent_parser2 = argparse.ArgumentParser(add_help=False)
         parent_parser2.add_argument('-p','-profile', '--profile',
             dest='Global.profile',type=str,
             default=NoConfigDefault,
             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')
         parent_parser2.add_argument('-cluster_dir', '--cluster-dir',
             dest='Global.cluster_dir',type=str,
             default=NoConfigDefault,
             help='Set the cluster dir. This overrides the logic used by the '
             '--profile option.',
             default=NoConfigDefault,
             metavar='Global.cluster_dir')
         parent_parser2.add_argument('--log-to-file',
             action='store_true', dest='Global.log_to_file', 
             default=NoConfigDefault,
             help='Log to a file in the log directory (default is stdout)'
         )
 
         subparsers = self.parser.add_subparsers(
             dest='Global.subcommand',
             title='ipcluster subcommands',
             description='ipcluster has a variety of subcommands. '
             'The general way of running ipcluster is "ipcluster <cmd> '
             ' [options]""',
             help='For more help, type "ipcluster <cmd> -h"')
 
         parser_list = subparsers.add_parser(
             'list',
             help='List all clusters in cwd and ipythondir.',
             parents=[parent_parser1]
         )
 
         parser_create = subparsers.add_parser(
             'create',
             help='Create a new cluster directory.',
             parents=[parent_parser1, parent_parser2] 
         )
         parser_create.add_argument(
             '--reset-config',
             dest='Global.reset_config', action='store_true',
             default=NoConfigDefault,
             help='Recopy the default config files to the cluster directory. '
             'You will loose any modifications you have made to these files.'
         )
 
         parser_start = subparsers.add_parser(
             'start',
             help='Start a cluster.',
             parents=[parent_parser1, parent_parser2]
         )
         parser_start.add_argument(
             '-n', '--number',
             type=int, dest='Global.n',
             default=NoConfigDefault,
             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'
 
 
 class IPClusterApp(ApplicationWithClusterDir):
 
     name = 'ipcluster'
     description = 'Start an IPython cluster (controller and engines).'
     config_file_name = default_config_file_name
     default_log_level = logging.INFO
     auto_create_cluster_dir = False
 
     def create_default_config(self):
         super(IPClusterApp, self).create_default_config()
         self.default_config.Global.controller_launcher = \
             '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."""
         return IPClusterCLLoader(
             description=self.description, 
             version=release.version
         )
 
     def find_resources(self):
         subcommand = self.command_line_config.Global.subcommand
         if subcommand=='list':
             self.list_cluster_dirs()
             # Exit immediately because there is nothing left to do.
             self.exit()
         elif subcommand=='create':
             self.auto_create_cluster_dir = True
             super(IPClusterApp, self).find_resources()
         elif subcommand=='start':
             self.auto_create_cluster_dir = False
             try:
                 super(IPClusterApp, self).find_resources()
             except ClusterDirError:
                 raise ClusterDirError(
                     "Could not find a cluster directory. A cluster dir must "
                     "be created before running 'ipcluster start'.  Do "
                     "'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':
             pass
         elif config.Global.subcommand=='create':
             self.log.info('Copying default config files to cluster directory '
             '[overwrite=%r]' % (config.Global.reset_config,))
             self.cluster_dir_obj.copy_all_config_files(overwrite=config.Global.reset_config)
         elif config.Global.subcommand=='start':
             self.start_logging()
             reactor.callWhenRunning(self.start_launchers)
 
     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)
             for f in files:
                 full_path = os.path.join(path, f)
                 if os.path.isdir(full_path) and f.startswith('cluster_'):
                     profile = full_path.split('_')[-1]
                     start_cmd = '"ipcluster start -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
 
         # Create the launchers
         el_class = import_item(config.Global.engine_launcher)
         self.engine_launcher = el_class(
             self.cluster_dir, config=config
         )
         cl_class = import_item(config.Global.controller_launcher)
         self.controller_launcher = cl_class(
             self.cluster_dir, config=config
         )
 
         # 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()
         d1.addCallback(self.stop_engines)
         d1.addErrback(self.err_and_stop)
         # If this triggers, just let them die
         # d2 = self.engine_launcher.observe_stop()
         
         # Start the controller and engines
         d = self.controller_launcher.start(
             profile=None, cluster_dir=config.Global.cluster_dir
         )
         d.addCallback(lambda _: self.start_engines())
         d.addErrback(self.err_and_stop)
 
     def err_and_stop(self, f):
         log.msg('Unexpected error in ipcluster:')
         log.err(f)
         reactor.stop()
 
     def stop_engines(self, r):
         return self.engine_launcher.stop()
 
     def start_engines(self):
         config = self.master_config
         d = self.engine_launcher.start(
             config.Global.n,
             profile=None, cluster_dir=config.Global.cluster_dir
         )
         return d
 
     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':
             return
         elif config.Global.subcommand=='start':
             reactor.run()
 
 
 def launch_new_instance():
     """Create and run the IPython cluster."""
     app = IPClusterApp()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
+

IPython/kernel/ipcontrollerapp.py

@@ -1,275 +1,254 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
     auto_create_cluster_dir = True
 
     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
+        self.default_config.Global.clean_logs = True
 
     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,249 +1,239 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
     auto_create_cluster_dir = True
 
     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
         # 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 = ''
 
+        # 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
         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)
-
+        super(IPEngineApp, self).pre_construct()
         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 not config.Global.furl_file:
             try_this = os.path.join(
                 config.Global.cluster_dir, 
                 config.Global.security_dir,
                 config.Global.furl_file_name
             )
             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
+            self.master_config.Global.furl_file,
+            self.master_config.Global.connect_delay,
+            self.master_config.Global.connect_max_tries
         )
 
         def handle_error(f):
             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/launcher.py

@@ -1,585 +1,628 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 Facilities for launching processing asynchronously.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import re
 import sys
 
 from IPython.core.component import Component
 from IPython.external import Itpl
 from IPython.utils.traitlets import Str, Int, List, Unicode
 from IPython.kernel.twistedutil import gatherBoth, make_deferred, sleep_deferred
 
 from twisted.internet import reactor, defer
 from twisted.internet.defer import inlineCallbacks
 from twisted.internet.protocol import ProcessProtocol
 from twisted.internet.utils import getProcessOutput
 from twisted.internet.error import ProcessDone, ProcessTerminated
 from twisted.python import log
 from twisted.python.failure import Failure
 
 #-----------------------------------------------------------------------------
 # Generic launchers
 #-----------------------------------------------------------------------------
 
 
 class LauncherError(Exception):
     pass
 
 
 class ProcessStateError(LauncherError):
     pass
 
 
 class UnknownStatus(LauncherError):
     pass
 
 
 class BaseLauncher(Component):
     """An asbtraction for starting, stopping and signaling a process."""
 
     working_dir = Unicode(u'')
 
     def __init__(self, working_dir, parent=None, name=None, config=None):
         super(BaseLauncher, self).__init__(parent, name, config)
         self.working_dir = working_dir
         self.state = 'before' # can be before, running, after
         self.stop_deferreds = []
         self.start_data = None
         self.stop_data = None
 
     @property
     def args(self):
         """A list of cmd and args that will be used to start the process."""
         return self.find_args()
 
     def find_args(self):
         """The ``.args`` property calls this to find the args list."""
         raise NotImplementedError('find_args must be implemented in a subclass')
 
     @property
     def arg_str(self):
         """The string form of the program arguments."""
         return ' '.join(self.args)
 
     @property
     def running(self):
         if self.state == 'running':
             return True
         else:
             return False
 
     def start(self):
         """Start the process.
 
         This must return a deferred that fires with information about the
         process starting (like a pid, job id, etc.)
         """
         return defer.fail(
             Failure(NotImplementedError(
                 'start must be implemented in a subclass')
             )
         )
 
     def stop(self):
         """Stop the process and notify observers of ProcessStopped.
 
         This must return a deferred that fires with any errors that occur
         while the process is attempting to be shut down.  This deferred
         won't fire when the process actually stops.  These events are 
         handled by calling :func:`observe_stop`.
         """
         return defer.fail(
             Failure(NotImplementedError(
                 'stop must be implemented in a subclass')
             )
         )
 
     def observe_stop(self):
         """Get a deferred that will fire when the process stops.
 
         The deferred will fire with data that contains information about
         the exit status of the process.
         """
         if self.state=='after':
             return defer.succeed(self.stop_data)
         else:
             d = defer.Deferred()
             self.stop_deferreds.append(d)
             return d
 
     def notify_start(self, data):
         """Call this to tigger startup actions.
 
         This logs the process startup and sets the state to running.  It is
         a pass-through so it can be used as a callback.
         """
 
         log.msg('Process %r started: %r' % (self.args[0], data))
         self.start_data = data
         self.state = 'running'
         return data
 
     def notify_stop(self, data):
         """Call this to trigger all the deferreds from :func:`observe_stop`."""
 
         log.msg('Process %r stopped: %r' % (self.args[0], data))
         self.stop_data = data
         self.state = 'after'
         for i in range(len(self.stop_deferreds)):
             d = self.stop_deferreds.pop()
             d.callback(data)
         return data
 
     def signal(self, sig):
         """Signal the process.
 
         Return a semi-meaningless deferred after signaling the process.
 
         Parameters
         ----------
         sig : str or int
             'KILL', 'INT', etc., or any signal number
         """
         return defer.fail(
             Failure(NotImplementedError(
                 'signal must be implemented in a subclass')
             )
         )
 
 
 class LocalProcessLauncherProtocol(ProcessProtocol):
     """A ProcessProtocol to go with the LocalProcessLauncher."""
 
     def __init__(self, process_launcher):
         self.process_launcher = process_launcher
         self.pid = None
 
     def connectionMade(self):
         self.pid = self.transport.pid
         self.process_launcher.notify_start(self.transport.pid)
 
     def processEnded(self, status):
         value = status.value
         if isinstance(value, ProcessDone):
             self.process_launcher.notify_stop(
                 {'exit_code':0,
                  'signal':None,
                  'status':None,
                  'pid':self.pid
                 }
             )
         elif isinstance(value, ProcessTerminated):
             self.process_launcher.notify_stop(
                 {'exit_code':value.exitCode,
                  'signal':value.signal,
                  'status':value.status,
                  'pid':self.pid
                 }
             )
         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 LocalProcessLauncher(BaseLauncher):
     """Start and stop an external process in an asynchronous manner."""
 
     cmd_and_args = List([])
 
     def __init__(self, working_dir, parent=None, name=None, config=None):
         super(LocalProcessLauncher, self).__init__(
             working_dir, parent, name, config
         )
         self.process_protocol = None
         self.start_deferred = None
 
     def find_args(self):
         return self.cmd_and_args
 
     def start(self):
         if self.state == 'before':
             self.process_protocol = LocalProcessLauncherProtocol(self)
             self.start_deferred = defer.Deferred()
             self.process_transport = reactor.spawnProcess(
                 self.process_protocol,
                 str(self.args[0]),
                 [str(a) for a in self.args],
                 env=os.environ
             )
             return self.start_deferred
         else:
             s = 'The process was already started and has state: %r' % self.state
             return defer.fail(ProcessStateError(s))
 
     def notify_start(self, data):
         super(LocalProcessLauncher, self).notify_start(data)
         self.start_deferred.callback(data)
 
     def stop(self):
         return self.interrupt_then_kill()
 
     @make_deferred
     def signal(self, sig):
         if self.state == 'running':
             self.process_transport.signalProcess(sig)
 
     @inlineCallbacks
     def interrupt_then_kill(self, delay=1.0):
         yield self.signal('INT')
         yield sleep_deferred(delay)
         yield self.signal('KILL')
 
 
 class MPIExecLauncher(LocalProcessLauncher):
 
     mpi_cmd = List(['mpiexec'], config=True)
     mpi_args = List([], config=True)
     program = List(['date'], config=True)
     program_args = List([], config=True)
     n = Int(1, config=True)
 
     def find_args(self):
         return self.mpi_cmd + ['-n', self.n] + self.mpi_args + \
                self.program + self.program_args
 
     def start(self, n):
         self.n = n
         return super(MPIExecLauncher, self).start()
 
 
 class SSHLauncher(BaseLauncher):
     """A minimal launcher for ssh.
 
     To be useful this will probably have to be extended to use the ``sshx``
     idea for environment variables.  There could be other things this needs
     as well.
     """
 
     ssh_cmd = List(['ssh'], config=True)
     ssh_args = List([], config=True)
     program = List(['date'], config=True)
     program_args = List([], config=True)
     hostname = Str('', config=True)
     user = Str(os.environ['USER'], config=True)
     location = Str('')
 
     def _hostname_changed(self, name, old, new):
         self.location = '%s@%s' % (self.user, new)
 
     def _user_changed(self, name, old, new):
         self.location = '%s@%s' % (new, self.hostname)
 
     def find_args(self):
         return self.ssh_cmd + self.ssh_args + [self.location] + \
                self.program + self.program_args
 
     def start(self, n, hostname=None, user=None):
         if hostname is not None:
             self.hostname = hostname
         if user is not None:
             self.user = user
         return super(SSHLauncher, self).start()
 
 
 class WindowsHPCLauncher(BaseLauncher):
     pass
 
 
 class BatchSystemLauncher(BaseLauncher):
 
     # Subclasses must fill these in.  See PBSEngineSet
     submit_command = Str('', config=True)
     delete_command = Str('', config=True)
     job_id_regexp = Str('', config=True)
     batch_template = Str('', config=True)
     batch_file_name = Unicode(u'batch_script', config=True)
     batch_file = Unicode(u'')
 
     def __init__(self, working_dir, parent=None, name=None, config=None):
         super(BatchSystemLauncher, self).__init__(
             working_dir, parent, name, config
         )
         self.batch_file = os.path.join(self.working_dir, self.batch_file_name)
         self.context = {}
 
     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 LauncherError("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
         script_as_string = Itpl.itplns(self.batch_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()
 
     @inlineCallbacks
     def start(self, n):
         """Start n copies of the process using a batch system."""
         self.write_batch_script(n)
         output = yield getProcessOutput(self.submit_command,
             [self.batch_file], env=os.environ)
         job_id = self.parse_job_id(output)
         self.notify_start(job_id)
         defer.returnValue(job_id)
 
     @inlineCallbacks
     def stop(self):
         output = yield getProcessOutput(self.delete_command,
             [self.job_id], env=os.environ
         )
         self.notify_stop(output)  # Pass the output of the kill cmd
         defer.returnValue(output)
 
 
 class PBSLauncher(BatchSystemLauncher):
 
     submit_command = Str('qsub', config=True)
     delete_command = Str('qdel', config=True)
     job_id_regexp = Str('\d+', config=True)
     batch_template = Str('', config=True)
     batch_file_name = Unicode(u'pbs_batch_script', config=True)
     batch_file = Unicode(u'')
 
 
 #-----------------------------------------------------------------------------
 # Controller launchers
 #-----------------------------------------------------------------------------
 
 def find_controller_cmd():
     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 import ipcontrollerapp
         script_location = ipcontrollerapp.__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 = ['ipcontroller']
     return cmd
 
 
 class LocalControllerLauncher(LocalProcessLauncher):
 
     controller_cmd = List(find_controller_cmd())
     controller_args = List(['--log-to-file','--log-level', '40'], config=True)
 
     def find_args(self):
         return self.controller_cmd + self.controller_args
 
     def start(self, profile=None, cluster_dir=None):
         if cluster_dir is not None:
             self.controller_args.extend(['--cluster-dir', cluster_dir])
         if profile is not None:
             self.controller_args.extend(['--profile', profile])
         log.msg("Starting LocalControllerLauncher: %r" % self.args)
         return super(LocalControllerLauncher, self).start()
 
 
 class WindowsHPCControllerLauncher(WindowsHPCLauncher):
     pass
 
 
 class MPIExecControllerLauncher(MPIExecLauncher):
 
     controller_cmd = List(find_controller_cmd(), config=False)
     controller_args = List(['--log-to-file','--log-level', '40'], config=True)
     n = Int(1, config=False)
 
     def start(self, profile=None, cluster_dir=None):
         if cluster_dir is not None:
             self.controller_args.extend(['--cluster-dir', cluster_dir])
         if profile is not None:
             self.controller_args.extend(['--profile', profile])
         log.msg("Starting MPIExecControllerLauncher: %r" % self.args)
         return super(MPIExecControllerLauncher, self).start(1)
 
 
     def find_args(self):
         return self.mpi_cmd + ['-n', self.n] + self.mpi_args + \
                self.controller_cmd + self.controller_args
 
 
 class PBSControllerLauncher(PBSLauncher):
 
     def start(self, profile=None, cluster_dir=None):
         # Here we save profile and cluster_dir in the context so they
         # can be used in the batch script template as ${profile} and
         # ${cluster_dir}
         if cluster_dir is not None:
             self.context['cluster_dir'] = cluster_dir
         if profile is not None:
             self.context['profile'] = profile
         log.msg("Starting PBSControllerLauncher: %r" % self.args)
         return super(PBSControllerLauncher, self).start(1)
 
 
 class SSHControllerLauncher(SSHLauncher):
     pass
 
 
 #-----------------------------------------------------------------------------
 # Engine launchers
 #-----------------------------------------------------------------------------
 
 
 def find_engine_cmd():
     if sys.platform == 'win32':
         # This logic is needed because the ipengine script doesn't
         # always get installed in the same way or in the same location.
         from IPython.kernel import ipengineapp
         script_location = ipengineapp.__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 = ['ipengine']
     return 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
 
     def start(self, profile=None, cluster_dir=None):
         if cluster_dir is not None:
             self.engine_args.extend(['--cluster-dir', cluster_dir])
         if profile is not None:
             self.engine_args.extend(['--profile', profile])
         return super(LocalEngineLauncher, self).start()
 
 
 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__(
             working_dir, parent, name, config
         )
         self.launchers = []
 
     def start(self, n, profile=None, cluster_dir=None):
         dlist = []
         for i in range(n):
             el = LocalEngineLauncher(self.working_dir, self)
             # Copy the engine args over to each engine launcher.
             import copy
             el.engine_args = copy.deepcopy(self.engine_args)
             d = el.start(profile, cluster_dir)
             if i==0:
                 log.msg("Starting LocalEngineSetLauncher: %r" % el.args)
             self.launchers.append(el)
             dlist.append(d)
         # The consumeErrors here could be dangerous
         dfinal = gatherBoth(dlist, consumeErrors=True)
         dfinal.addCallback(self.notify_start)
         return dfinal
 
     def find_args(self):
         return ['engine set']
 
     def signal(self, sig):
         dlist = []
         for el in self.launchers:
             d = el.signal(sig)
             dlist.append(d)
         dfinal = gatherBoth(dlist, consumeErrors=True)
         return dfinal
 
     def interrupt_then_kill(self, delay=1.0):
         dlist = []
         for el in self.launchers:
             d = el.interrupt_then_kill(delay)
             dlist.append(d)
         dfinal = gatherBoth(dlist, consumeErrors=True)
         return dfinal
 
     def stop(self):
         return self.interrupt_then_kill()
 
     def observe_stop(self):
         dlist = [el.observe_stop() for el in self.launchers]
         dfinal = gatherBoth(dlist, consumeErrors=False)
         dfinal.addCallback(self.notify_stop)
         return dfinal
 
 
 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):
         if cluster_dir is not None:
             self.engine_args.extend(['--cluster-dir', cluster_dir])
         if profile is not None:
             self.engine_args.extend(['--profile', profile])
         log.msg('Starting MPIExecEngineSetLauncher: %r' % self.args)
         return super(MPIExecEngineSetLauncher, self).start(n)
 
     def find_args(self):
         return self.mpi_cmd + ['-n', self.n] + self.mpi_args + \
                self.engine_cmd + self.engine_args
 
 
 class WindowsHPCEngineSetLauncher(WindowsHPCLauncher):
     pass
 
 
 class PBSEngineSetLauncher(PBSLauncher):
 
     def start(self, n, profile=None, cluster_dir=None):
         if cluster_dir is not None:
             self.program_args.extend(['--cluster-dir', cluster_dir])
         if profile is not None:
             self.program_args.extend(['-p', profile])
         log.msg('Starting PBSEngineSetLauncher: %r' % self.args)
         return super(PBSEngineSetLauncher, self).start(n)
 
 
 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

@@ -1,265 +1,263 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
+#-----------------------------------------------------------------------------
+#  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
 
 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
 
 
 def make_deferred(func):
     """A decorator that calls a function with :func`maybeDeferred`."""
 
     def _wrapper(*args, **kwargs):
         return defer.maybeDeferred(func, *args, **kwargs)
 
     return _wrapper
  No newline at end of file