Lots more work on the kernel scripts.

IPython/config/default/ipengine_config.py

@@ -1,32 +1,24 b''
 c = get_config()
 
-c.MPI.use = ''
+# c.Global.log_to_file = 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 = ''
+
+# 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
+# """
 
-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
-"""
-
-c.Global.log_to_file = 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 = ''

IPython/core/application.py

@@ -1,345 +1,364 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 An application for IPython.
 
 All top-level applications should use the classes in this module for
 handling configuration and creating componenets.
 
 The job of an :class:`Application` is to create the master configuration 
 object and then create the components, passing the config to them.
 
 Authors:
 
 * Brian Granger
 * Fernando Perez
 
 Notes
 -----
 """
 
 #-----------------------------------------------------------------------------
 #  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 sys
 
 from IPython.core import release
 from IPython.utils.genutils import get_ipython_dir
 from IPython.config.loader import (
     PyFileConfigLoader,
     ArgParseConfigLoader,
     Config,
     NoConfigDefault
 )
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 
 class BaseAppArgParseConfigLoader(ArgParseConfigLoader):
     """Default command line options for IPython based applications."""
 
     def _add_other_arguments(self):
         self.parser.add_argument('-ipythondir', '--ipython-dir', 
             dest='Global.ipythondir',type=str,
             help='Set to override default location of Global.ipythondir.',
             default=NoConfigDefault,
             metavar='Global.ipythondir')
         self.parser.add_argument('-p','-profile', '--profile',
             dest='Global.profile',type=str,
             help='The string name of the ipython profile to be used.',
             default=NoConfigDefault,
             metavar='Global.profile')
         self.parser.add_argument('-log_level', '--log-level',
             dest="Global.log_level",type=int,
             help='Set the log level (0,10,20,30,40,50).  Default is 30.',
             default=NoConfigDefault,
             metavar='Global.log_level')
         self.parser.add_argument('-config_file', '--config-file',
             dest='Global.config_file',type=str,
             help='Set the config file name to override default.',
             default=NoConfigDefault,
             metavar='Global.config_file')
 
 
 class ApplicationError(Exception):
     pass
 
 
 class Application(object):
     """Load a config, construct components and set them running."""
 
     name = 'ipython'
     description = 'IPython: an enhanced interactive Python shell.'
     config_file_name = 'ipython_config.py'
     default_log_level = logging.WARN
 
     def __init__(self):
+        self._exiting = False
         self.init_logger()
         # Track the default and actual separately because some messages are
         # only printed if we aren't using the default.
         self.default_config_file_name = self.config_file_name
 
     def init_logger(self):
         self.log = logging.getLogger(self.__class__.__name__)
         # This is used as the default until the command line arguments are read.
         self.log.setLevel(self.default_log_level)
         self._log_handler = logging.StreamHandler()
         self._log_formatter = logging.Formatter("[%(name)s] %(message)s")
         self._log_handler.setFormatter(self._log_formatter)
         self.log.addHandler(self._log_handler)
 
     def _set_log_level(self, level):
         self.log.setLevel(level)
 
     def _get_log_level(self):
         return self.log.level
 
     log_level = property(_get_log_level, _set_log_level)
 
     def start(self):
         """Start the application."""
         self.attempt(self.create_default_config)
         self.log_default_config()
         self.set_default_config_log_level()
         self.attempt(self.pre_load_command_line_config)
         self.attempt(self.load_command_line_config, action='abort')
         self.set_command_line_config_log_level()
         self.attempt(self.post_load_command_line_config)
         self.log_command_line_config()
         self.attempt(self.find_ipythondir)
+        self.attempt(self.find_resources)
         self.attempt(self.find_config_file_name)
         self.attempt(self.find_config_file_paths)
         self.attempt(self.pre_load_file_config)
         self.attempt(self.load_file_config)
         self.set_file_config_log_level()
         self.attempt(self.post_load_file_config)
         self.log_file_config()
         self.attempt(self.merge_configs)
         self.log_master_config()
         self.attempt(self.pre_construct)
         self.attempt(self.construct)
         self.attempt(self.post_construct)
         self.attempt(self.start_app)
 
     #-------------------------------------------------------------------------
     # Various stages of Application creation
     #-------------------------------------------------------------------------
 
     def create_default_config(self):
         """Create defaults that can't be set elsewhere.
 
         For the most part, we try to set default in the class attributes
         of Components.  But, defaults the top-level Application (which is
         not a HasTraitlets or Component) are not set in this way.  Instead
         we set them here.  The Global section is for variables like this that
         don't belong to a particular component.
         """
         self.default_config = Config()
         self.default_config.Global.ipythondir = get_ipython_dir()
         self.default_config.Global.log_level = self.log_level
 
     def log_default_config(self):
         self.log.debug('Default config loaded:')
         self.log.debug(repr(self.default_config))
 
     def set_default_config_log_level(self):
         try:
             self.log_level = self.default_config.Global.log_level
         except AttributeError:
             # Fallback to the default_log_level class attribute
             pass
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return BaseAppArgParseConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def pre_load_command_line_config(self):
         """Do actions just before loading the command line config."""
         pass
 
     def load_command_line_config(self):
         """Load the command line config."""
         loader = self.create_command_line_config()
         self.command_line_config = loader.load_config()
         self.extra_args = loader.get_extra_args()
 
     def set_command_line_config_log_level(self):
         try:
             self.log_level = self.command_line_config.Global.log_level
         except AttributeError:
             pass
 
     def post_load_command_line_config(self):
         """Do actions just after loading the command line config."""
         pass
 
     def log_command_line_config(self):
         self.log.debug("Command line config loaded:")
         self.log.debug(repr(self.command_line_config))
 
     def find_ipythondir(self):
         """Set the IPython directory.
 
         This sets ``self.ipythondir``, but the actual value that is passed
         to the application is kept in either ``self.default_config`` or
         ``self.command_line_config``.  This also adds ``self.ipythondir`` to
         ``sys.path`` so config files there can be references by other config
         files.
         """
 
         try:
             self.ipythondir = self.command_line_config.Global.ipythondir
         except AttributeError:
             self.ipythondir = self.default_config.Global.ipythondir
         sys.path.append(os.path.abspath(self.ipythondir))
         if not os.path.isdir(self.ipythondir):
             os.makedirs(self.ipythondir, mode=0777)
         self.log.debug("IPYTHONDIR set to: %s" % self.ipythondir)
 
+    def find_resources(self):
+        """Find other resources that need to be in place.
+
+        Things like cluster directories need to be in place to find the
+        config file.  These happen right after the IPython directory has
+        been set.
+        """
+        pass
+
     def find_config_file_name(self):
         """Find the config file name for this application.
 
         This must set ``self.config_file_name`` to the filename of the
         config file to use (just the filename). The search paths for the
         config file are set in :meth:`find_config_file_paths` and then passed
         to the config file loader where they are resolved to an absolute path.
 
         If a profile has been set at the command line, this will resolve
         it.
         """
 
         try:
             self.config_file_name = self.command_line_config.Global.config_file
         except AttributeError:
             pass
 
         try:
             self.profile_name = self.command_line_config.Global.profile
             name_parts = self.config_file_name.split('.')
             name_parts.insert(1, '_' + self.profile_name + '.')
             self.config_file_name = ''.join(name_parts)
         except AttributeError:
             pass
 
     def find_config_file_paths(self):
         """Set the search paths for resolving the config file.
 
         This must set ``self.config_file_paths`` to a sequence of search
         paths to pass to the config file loader.
         """
         self.config_file_paths = (os.getcwd(), self.ipythondir)
 
     def pre_load_file_config(self):
         """Do actions before the config file is loaded."""
         pass
 
     def load_file_config(self):
         """Load the config file.
         
         This tries to load the config file from disk.  If successful, the
         ``CONFIG_FILE`` config variable is set to the resolved config file
         location.  If not successful, an empty config is used.
         """
         self.log.debug("Attempting to load config file: %s" % self.config_file_name)
         loader = PyFileConfigLoader(self.config_file_name,
                                     path=self.config_file_paths)
         try:
             self.file_config = loader.load_config()
             self.file_config.Global.config_file = loader.full_filename
         except IOError:
             # Only warn if the default config file was NOT being used.
             if not self.config_file_name==self.default_config_file_name:
                 self.log.warn("Config file not found, skipping: %s" % \
                                self.config_file_name, exc_info=True)
             self.file_config = Config()
         except:
             self.log.warn("Error loading config file: %s" % \
                            self.config_file_name, exc_info=True)
             self.file_config = Config()
 
     def set_file_config_log_level(self):
         # We need to keeep self.log_level updated.  But we only use the value
         # of the file_config if a value was not specified at the command
         # line, because the command line overrides everything.
         if not hasattr(self.command_line_config.Global, 'log_level'):
             try:
                 self.log_level = self.file_config.Global.log_level
             except AttributeError:
                 pass # Use existing value
 
     def post_load_file_config(self):
         """Do actions after the config file is loaded."""
         pass
 
     def log_file_config(self):
         if hasattr(self.file_config.Global, 'config_file'):
             self.log.debug("Config file loaded: %s" % self.file_config.Global.config_file)
             self.log.debug(repr(self.file_config))
 
     def merge_configs(self):
         """Merge the default, command line and file config objects."""
         config = Config()
         config._merge(self.default_config)
         config._merge(self.file_config)
         config._merge(self.command_line_config)
         self.master_config = config
 
     def log_master_config(self):
         self.log.debug("Master config created:")
         self.log.debug(repr(self.master_config))
 
     def pre_construct(self):
         """Do actions after the config has been built, but before construct."""
         pass
 
     def construct(self):
         """Construct the main components that make up this app."""
         self.log.debug("Constructing components for application")
 
     def post_construct(self):
         """Do actions after construct, but before starting the app."""
         pass
 
     def start_app(self):
         """Actually start the app."""
         self.log.debug("Starting application")
 
     #-------------------------------------------------------------------------
     # Utility methods
     #-------------------------------------------------------------------------
 
     def abort(self):
         """Abort the starting of the application."""
-        self.log.critical("Aborting application: %s" % self.name, exc_info=True)
-        sys.exit(1)
+        if self._exiting:
+            pass
+        else:
+            self.log.critical("Aborting application: %s" % self.name, exc_info=True)
+            self._exiting = True
+            sys.exit(1)
 
     def exit(self):
-        self.log.critical("Aborting application: %s" % self.name)
-        sys.exit(1)
+        if self._exiting:
+            pass
+        else:
+            self.log.debug("Exiting application: %s" % self.name)
+            self._exiting = True
+            sys.exit(1)
 
     def attempt(self, func, action='abort'):
         try:
             func()
         except SystemExit:
-            self.exit()
+            raise
         except:
             if action == 'abort':
                 self.abort()
             elif action == 'exit':
                 self.exit()
 

IPython/kernel/clusterdir.py

@@ -1,263 +1,352 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython cluster directory
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import shutil
 
 from IPython.core import release
 from IPython.config.loader import PyFileConfigLoader
 from IPython.core.application import Application
 from IPython.core.component import Component
 from IPython.config.loader import ArgParseConfigLoader, NoConfigDefault
-from IPython.utils.traitlets import Unicode
+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 find_cluster_dir_by_profile(cls, path, profile='default'):
-        """Find/create a cluster dir by profile name and return its ClusterDir.
+    def create_cluster_dir(csl, cluster_dir):
+        """Create a new cluster directory given a full path.
 
-        This will create the cluster directory if it doesn't exist.
+        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
         ----------
-        path : unicode or str
-            The directory path to look for the cluster directory in.
+        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 = os.path.join(os.getcwd(), dirname)
-        if os.path.isdir(cluster_dir):
-            return ClusterDir(cluster_dir)
+        cluster_dir_paths = os.environ.get('IPCLUSTERDIR_PATH','')
+        if cluster_dir_paths:
+            cluster_dir_paths = cluster_dir_paths.split(':')
         else:
-            if not os.path.isdir(path):
-                raise IOError("Directory doesn't exist: %s" % path)
-            cluster_dir = os.path.join(path, dirname)
-            return ClusterDir(cluster_dir)
+            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')
         self.parser.add_argument('-p','-profile', '--profile',
             dest='Global.profile',type=str,
             help='The string name of the profile to be used. This determines '
             'the name of the cluster dir as: cluster_<profile>. The default profile '
             'is named "default".  The cluster directory is resolve this way '
             'if the --cluster-dir option is not used.',
             default=NoConfigDefault,
             metavar='Global.profile')
         self.parser.add_argument('-log_level', '--log-level',
             dest="Global.log_level",type=int,
             help='Set the log level (0,10,20,30,40,50).  Default is 30.',
-            default=NoConfigDefault)
+            default=NoConfigDefault,
+            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')
 
 
 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 = ''
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return AppWithClusterDirArgParseConfigLoader(
             description=self.description, 
             version=release.version
         )
 
-    def find_config_file_name(self):
-        """Find the config file name for this application."""
-        # For this type of Application it should be set as a class attribute.
-        if not hasattr(self, 'config_file_name'):
-            self.log.critical("No config filename found")
-
-    def find_config_file_paths(self):
-        """This resolves the cluster directory and sets ``config_file_paths``.
+    def find_resources(self):
+        """This resolves the cluster directory.
 
-        This does the following:
-        * Create the :class:`ClusterDir` object for the application.
-        * Set the ``cluster_dir`` attribute of the application and config
+        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.
-        * Set ``config_file_paths`` to point to the cluster directory.
+
+        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`.
         """
 
-        # Create the ClusterDir object for managing everything
         try:
             cluster_dir = self.command_line_config.Global.cluster_dir
         except AttributeError:
             cluster_dir = self.default_config.Global.cluster_dir
         cluster_dir = os.path.expandvars(os.path.expanduser(cluster_dir))
-        if cluster_dir:
-            # Just use cluster_dir
+        try:
             self.cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
+        except ClusterDirError:
+            pass
         else:
-            # Then look for a profile
-            try:
-                self.profile = self.command_line_config.Global.profile
-            except AttributeError:
-                self.profile = self.default_config.Global.profile
+            self.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,)
+
+

IPython/kernel/ipcontrollerapp.py

@@ -1,274 +1,275 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
 
     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,248 +1,249 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()
 
         # Global config attributes
         self.default_config.Global.log_to_file = False
         self.default_config.Global.exec_lines = []
         # The log and security dir names must match that of the controller
         self.default_config.Global.log_dir_name = 'log'
         self.default_config.Global.security_dir_name = 'security'
         self.default_config.Global.shell_class = 'IPython.kernel.core.interpreter.Interpreter'
 
         # Configuration related to the controller
         # This must match the filename (path not included) that the controller
         # used for the FURL file.
         self.default_config.Global.furl_file_name = 'ipcontroller-engine.furl'
         # If given, this is the actual location of the controller's FURL file.
         # If not, this is computed using the profile, app_dir and furl_file_name
         self.default_config.Global.furl_file = ''
 
         # MPI related config attributes
         self.default_config.MPI.use = ''
         self.default_config.MPI.mpi4py = mpi4py_init
         self.default_config.MPI.pytrilinos = pytrilinos_init
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return IPEngineAppCLConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def post_load_command_line_config(self):
         pass
 
     def pre_construct(self):
         config = self.master_config
         sdir = self.cluster_dir_obj.security_dir
         self.security_dir = config.Global.security_dir = sdir
         ldir = self.cluster_dir_obj.log_dir
         self.log_dir = config.Global.log_dir = ldir
         self.log.info("Cluster directory set to: %s" % self.cluster_dir)
         self.log.info("Log directory set to: %s" % self.log_dir)
         self.log.info("Security directory set to: %s" % self.security_dir)
 
         self.find_cont_furl_file()
 
     def find_cont_furl_file(self):
         """Set the furl file.
 
         Here we don't try to actually see if it exists for is valid as that
         is hadled by the connection logic.
         """
         config = self.master_config
         # Find the actual controller FURL file
         if 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
         )
 
         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/twistedutil.py

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