Minor improvements to the parallel computing stuff.

* Made various user messages better.
* Added a longer (4s) paused before ipcluster exits.

IPython/kernel/clientconnector.py

@@ -1,766 +1,773 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 """Facilities for handling client connections to the controller."""
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from __future__ import with_statement
 import os
 
 from IPython.kernel.fcutil import (
     Tub,
     find_furl,
     is_valid_furl_or_file,
     validate_furl_or_file,
     FURLError
 )
 from IPython.kernel.clusterdir import ClusterDir, ClusterDirError
 from IPython.kernel.launcher import IPClusterLauncher
 from IPython.kernel.twistedutil import (
     gatherBoth, 
     make_deferred,
     blockingCallFromThread, 
     sleep_deferred
 )
 from IPython.utils.importstring import import_item
 from IPython.utils.genutils import get_ipython_dir
 
 from twisted.internet import defer
 from twisted.internet.defer import inlineCallbacks, returnValue
 from twisted.python import failure, log
 
 #-----------------------------------------------------------------------------
 # The ClientConnector class
 #-----------------------------------------------------------------------------
 
 DELAY = 0.2
 MAX_TRIES = 9
 
 
 class ClientConnectorError(Exception):
     pass
 
 
 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 _find_furl(self, profile='default', cluster_dir=None, 
                    furl_or_file=None, furl_file_name=None,
                    ipython_dir=None):
         """Find a FURL file by profile+ipython_dir or cluster dir.
 
         This raises an :exc:`~IPython.kernel.fcutil.FURLError` exception 
         if a FURL file can't be found.
         """
         # Try by furl_or_file
         if furl_or_file is not None:
             validate_furl_or_file(furl_or_file)
             return furl_or_file
 
         if furl_file_name is None:
             raise FURLError('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)
             validate_furl_or_file(furl_file)
             return furl_file
 
         # Try by profile
         if ipython_dir is None:
             ipython_dir = get_ipython_dir()
         if profile is not None:
             cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
                 ipython_dir, profile)
             sdir = cluster_dir_obj.security_dir
             furl_file = os.path.join(sdir, furl_file_name)
             validate_furl_or_file(furl_file)
             return furl_file
 
         raise FURLError('Could not find a valid FURL file.')
 
     def get_reference(self, furl_or_file):
         """Get a remote reference using a furl or a file containing a furl.
 
         Remote references are cached locally so once a remote reference
         has been retrieved for a given furl, the cached version is 
         returned.
 
         Parameters
         ----------
         furl_or_file : str
             A furl or a filename containing a furl. This should already be
             validated, but might not yet exist.
 
         Returns
         -------
         A deferred to a remote reference
         """
         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)
         return d
         
     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, profile='default', cluster_dir=None,
                         furl_or_file=None, ipython_dir=None,
                         delay=DELAY, max_tries=MAX_TRIES):
         """Get the task controller client.
         
         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
         ----------
         profile : str
             The name of a cluster directory profile (default="default"). The
             cluster directory "cluster_<profile>" will be searched for
             in ``os.getcwd()``, the ipython_dir and then in the directories
             listed in the :env:`IPCLUSTER_DIR_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.
         ipython_dir : str
             The location of the ipython_dir if different from the default.
             This is used if the cluster directory is being found by profile.
         delay : float
             The initial delay between re-connection attempts. Susequent delays
             get longer according to ``delay[i] = 1.5*delay[i-1]``.
         max_tries : int
             The max number of re-connection attempts.
 
         Returns
         -------
         A deferred to the actual client class.
         """
         return self.get_client(
             profile, cluster_dir, furl_or_file, 
             'ipcontroller-tc.furl', ipython_dir,
             delay, max_tries
         )
 
     def get_multiengine_client(self, profile='default', cluster_dir=None,
                                furl_or_file=None, ipython_dir=None,
                                delay=DELAY, max_tries=MAX_TRIES):
         """Get the multiengine controller client.
         
         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
         ----------
         profile : str
             The name of a cluster directory profile (default="default"). The
             cluster directory "cluster_<profile>" will be searched for
             in ``os.getcwd()``, the ipython_dir and then in the directories
             listed in the :env:`IPCLUSTER_DIR_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.
         ipython_dir : str
             The location of the ipython_dir if different from the default.
             This is used if the cluster directory is being found by profile.
         delay : float
             The initial delay between re-connection attempts. Susequent delays
             get longer according to ``delay[i] = 1.5*delay[i-1]``.
         max_tries : int
             The max number of re-connection attempts.
 
         Returns
         -------
         A deferred to the actual client class.
         """
         return self.get_client(
             profile, cluster_dir, furl_or_file, 
             'ipcontroller-mec.furl', ipython_dir,
             delay, max_tries
         )
     
     def get_client(self, profile='default', cluster_dir=None,
                    furl_or_file=None, furl_file_name=None, ipython_dir=None,
                    delay=DELAY, max_tries=MAX_TRIES):
         """Get a remote reference and wrap it in a client by furl.
 
         This method is a simple wrapper around `get_client` that passes in
         the default name of the task client FURL file.  Usually only
         the ``profile`` option will be needed.  If a FURL file can't be
         found by its profile, use ``cluster_dir`` or ``furl_or_file``.
         
         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 ipython_dir and then in the directories
             listed in the :env:`IPCLUSTER_DIR_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. 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.
         ipython_dir : str
             The location of the ipython_dir if different from the default.
             This is used if the cluster directory is being found by profile.
         delay : float
             The initial delay between re-connection attempts. Susequent delays
             get longer according to ``delay[i] = 1.5*delay[i-1]``.
         max_tries : int
             The max number of re-connection attempts.
 
         Returns
         -------
         A deferred to the actual client class. Or a failure to a 
         :exc:`FURLError`.
         """
         try:
             furl_file = self._find_furl(
                 profile, cluster_dir, furl_or_file,
                 furl_file_name, ipython_dir
             )
         except FURLError:
             return defer.fail(failure.Failure())
 
         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 = self._try_to_connect(furl_file, delay, max_tries, attempt=0)
         d.addCallback(_wrap_remote_reference)
+        d.addErrback(self._handle_error, furl_file)
         return d
 
+    def _handle_error(self, f, furl_file):
+        raise ClientConnectorError('Could not connect to the controller '
+            'using the FURL file. This usually means that i) the controller '
+            'was not started or ii) a firewall was blocking the client from '
+            'connecting to the controller: %s' % furl_file)
+
     @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("Connecting to controller [%r]: %s" % \
-                (attempt, furl_or_file))
+            log.msg("Connecting [%r]" % attempt)
             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.get_reference(self.furl)
+                log.msg("Connected: %s" % furl_or_file)
             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)
                     rr = yield self._try_to_connect(
                         furl_or_file, 1.5*delay, max_tries, attempt+1
                     )
                     returnValue(rr)
             else:
                 returnValue(rr)
         else:
             raise ClientConnectorError(
                 '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 client from connecting '
                 'to the controller.' % max_tries
             )
 
 
 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, ipython_dir=None,
                         delay=DELAY, max_tries=MAX_TRIES):
         """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 ipython_dir and then in the directories
             listed in the :env:`IPCLUSTER_DIR_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.
         ipython_dir : str
             The location of the ipython_dir if different from the default.
             This is used if the cluster directory is being found by profile.
         delay : float
             The initial delay between re-connection attempts. Susequent delays
             get longer according to ``delay[i] = 1.5*delay[i-1]``.
         max_tries : int
             The max number of re-connection attempts.
 
         Returns
         -------
         The task client instance.
         """
         client = blockingCallFromThread(
             self.async_cc.get_task_client, profile, cluster_dir,
             furl_or_file, ipython_dir, delay, max_tries
         )
         return client.adapt_to_blocking_client()
 
     def get_multiengine_client(self, profile='default', cluster_dir=None,
                                furl_or_file=None, ipython_dir=None,
                                delay=DELAY, max_tries=MAX_TRIES):
         """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 ipython_dir and then in the directories
             listed in the :env:`IPCLUSTER_DIR_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.
         ipython_dir : str
             The location of the ipython_dir if different from the default.
             This is used if the cluster directory is being found by profile.
         delay : float
             The initial delay between re-connection attempts. Susequent delays
             get longer according to ``delay[i] = 1.5*delay[i-1]``.
         max_tries : int
             The max number of re-connection attempts.
 
         Returns
         -------
         The multiengine client instance.
         """
         client = blockingCallFromThread(
             self.async_cc.get_multiengine_client, profile, cluster_dir,
             furl_or_file, ipython_dir, delay, max_tries
         )
         return client.adapt_to_blocking_client()
 
     def get_client(self, profile='default', cluster_dir=None,
                    furl_or_file=None, ipython_dir=None,
                    delay=DELAY, max_tries=MAX_TRIES):
         client = blockingCallFromThread(
             self.async_cc.get_client, profile, cluster_dir,
             furl_or_file, ipython_dir,
             delay, max_tries
         )
         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, ipython_dir=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 ipython_dir and then in the directories
             listed in the :env:`IPCLUSTER_DIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
         ipython_dir : str
             The location of the ipython_dir 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, ipython_dir, 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, ipython_dir, auto_create):
         if ipython_dir is None:
             ipython_dir = 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(
                     ipython_dir, 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(
                 ipython_dir, 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, delay=DELAY, max_tries=MAX_TRIES):
         """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,
             delay=delay, max_tries=max_tries
         )
 
     def get_task_client(self, delay=DELAY, max_tries=MAX_TRIES):
         """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,
             delay=delay, max_tries=max_tries
         )
 
     def get_ipengine_logs(self):
         return self.get_logs_by_name('ipengine')
 
     def get_ipcontroller_logs(self):
         return self.get_logs_by_name('ipcontroller')
 
     def get_ipcluster_logs(self):
         return self.get_logs_by_name('ipcluster')
 
     def get_logs_by_name(self, name='ipcluster'):
         log_dir = self.cluster_dir_obj.log_dir
         logs = {}
         for log in os.listdir(log_dir):
             if log.startswith(name + '-') and log.endswith('.log'):
                 with open(os.path.join(log_dir, log), 'r') as f:
                     logs[log] = f.read()
         return logs
 
     def get_logs(self):
         d = self.get_ipcluster_logs()
         d.update(self.get_ipengine_logs())
         d.update(self.get_ipcontroller_logs())
         return d
 
     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, ipython_dir=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 ipython_dir and then in the directories
             listed in the :env:`IPCLUSTER_DIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
         ipython_dir : str
             The location of the ipython_dir 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, ipython_dir, 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, delay=DELAY, max_tries=MAX_TRIES):
         """Get the multiengine client for the running cluster.
 
         This will try to attempt to the controller multiple times. If this 
         fails altogether, try looking at the following:
         * Make sure the controller is starting properly by looking at its
           log files.
         * Make sure the controller is writing its FURL file in the location
           expected by the client.
         * Make sure a firewall on the controller's host is not blocking the
           client from connecting.
 
         Parameters
         ----------
         delay : float
             The initial delay between re-connection attempts. Susequent delays
             get longer according to ``delay[i] = 1.5*delay[i-1]``.
         max_tries : int
             The max number of re-connection attempts.
         """
         if self.client_connector is None:
             self.client_connector = ClientConnector()
         return self.client_connector.get_multiengine_client(
             cluster_dir=self.cluster_dir_obj.location,
             delay=delay, max_tries=max_tries
         )
 
     def get_task_client(self, delay=DELAY, max_tries=MAX_TRIES):
         """Get the task client for the running cluster.
 
         This will try to attempt to the controller multiple times. If this 
         fails altogether, try looking at the following:
         * Make sure the controller is starting properly by looking at its
           log files.
         * Make sure the controller is writing its FURL file in the location
           expected by the client.
         * Make sure a firewall on the controller's host is not blocking the
           client from connecting.
 
         Parameters
         ----------
         delay : float
             The initial delay between re-connection attempts. Susequent delays
             get longer according to ``delay[i] = 1.5*delay[i-1]``.
         max_tries : int
             The max number of re-connection attempts.
         """
         if self.client_connector is None:
             self.client_connector = ClientConnector()
         return self.client_connector.get_task_client(
             cluster_dir=self.cluster_dir_obj.location,
             delay=delay, max_tries=max_tries
         )
 
     def __repr__(self):
         s = "<Cluster(running=%r, location=%s)" % (self.running, self.location)
         return s
 
     def get_logs_by_name(self, name='ipcluter'):
         """Get a dict of logs by process name (ipcluster, ipengine, etc.)"""
         return self.async_cluster.get_logs_by_name(name)
 
     def get_ipengine_logs(self):
         """Get a dict of logs for all engines in this cluster."""
         return self.async_cluster.get_ipengine_logs()
 
     def get_ipcontroller_logs(self):
         """Get a dict of logs for the controller in this cluster."""
         return self.async_cluster.get_ipcontroller_logs()
 
     def get_ipcluster_logs(self):
         """Get a dict of the ipcluster logs for this cluster."""
         return self.async_cluster.get_ipcluster_logs()
 
     def get_logs(self):
         """Get a dict of all logs for this cluster."""
         return self.async_cluster.get_logs()
 
     def _print_logs(self, logs):
         for k, v in logs.iteritems():
             print "==================================="
             print "Logfile: %s" % k
             print "==================================="
             print v
             print
 
     def print_ipengine_logs(self):
         """Print the ipengine logs for this cluster to stdout."""
         self._print_logs(self.get_ipengine_logs())
 
     def print_ipcontroller_logs(self):
         """Print the ipcontroller logs for this cluster to stdout."""
         self._print_logs(self.get_ipcontroller_logs())
 
     def print_ipcluster_logs(self):
         """Print the ipcluster logs for this cluster to stdout."""
         self._print_logs(self.get_ipcluster_logs())
 
     def print_logs(self):
         """Print all the logs for this cluster to stdout."""
         self._print_logs(self.get_logs())
 

IPython/kernel/ipclusterapp.py

@@ -1,457 +1,457 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
 
 if os.name=='posix':
     from twisted.scripts._twistd_unix import daemonize
 
 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, PIDFileError
 )
 
 from twisted.internet import reactor, defer
 from twisted.python import log, failure
 
 
 #-----------------------------------------------------------------------------
 # The ipcluster application
 #-----------------------------------------------------------------------------
 
 
 # Exit codes for ipcluster
 
 # This will be the exit code if the ipcluster appears to be running because
 # a .pid file exists
 ALREADY_STARTED = 10
 
 # This will be the exit code if ipcluster stop is run, but there is not .pid
 # file to be found.
 ALREADY_STOPPED = 11
 
 
 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('--ipython-dir', 
             dest='Global.ipython_dir',type=unicode,
             help='Set to override default location of Global.ipython_dir.',
             default=NoConfigDefault,
             metavar='Global.ipython_dir')
         parent_parser1.add_argument('--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',
             dest='Global.profile',type=unicode,
             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',
             dest='Global.cluster_dir',type=unicode,
             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('--work-dir',
             dest='Global.work_dir',type=unicode,
             help='Set the working dir for the process.',
             default=NoConfigDefault,
             metavar='Global.work_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 ipython_dir.',
             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',
             dest='Global.clean_logs', action='store_true',
             help='Delete old log flies before starting.',
             default=NoConfigDefault
         )
         parser_start.add_argument('--no-clean-logs',
             dest='Global.clean_logs', action='store_false',
             help="Don't delete old log flies before starting.",
             default=NoConfigDefault
         )
         parser_start.add_argument('--daemon',
             dest='Global.daemonize', action='store_true',
             help='Daemonize the ipcluster program. This implies --log-to-file',
             default=NoConfigDefault
         )
         parser_start.add_argument('--no-daemon',
             dest='Global.daemonize', action='store_false',
             help="Dont't daemonize the ipcluster program.",
             default=NoConfigDefault
         )
 
         parser_start = subparsers.add_parser(
             'stop',
             help='Stop a cluster.',
             parents=[parent_parser1, parent_parser2]
         )
         parser_start.add_argument('--signal',
             dest='Global.signal', type=int,
             help="The signal number to use in stopping the cluster (default=2).",
             metavar="Global.signal",
             default=NoConfigDefault
         )
 
 
 default_config_file_name = u'ipcluster_config.py'
 
 
 class IPClusterApp(ApplicationWithClusterDir):
 
     name = u'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.n = 2
         self.default_config.Global.reset_config = False
         self.default_config.Global.clean_logs = True
         self.default_config.Global.signal = 2
         self.default_config.Global.daemonize = False
 
     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' or subcommand=='stop':
             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 list_cluster_dirs(self):
         # Find the search paths
         cluster_dir_paths = os.environ.get('IPCLUSTER_DIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
         try:
             ipython_dir = self.command_line_config.Global.ipython_dir
         except AttributeError:
             ipython_dir = self.default_config.Global.ipython_dir
         paths = [os.getcwd(), ipython_dir] + \
             cluster_dir_paths
         paths = list(set(paths))
 
         self.log.info('Searching for cluster dirs in paths: %r' % paths)
         for path in paths:
             files = os.listdir(path)
             for f in files:
                 full_path = os.path.join(path, f)
                 if os.path.isdir(full_path) and f.startswith('cluster_'):
                     profile = full_path.split('_')[-1]
-                    start_cmd = '"ipcluster start -n 4 -p %s"' % profile
+                    start_cmd = 'ipcluster start -p %s -n 4' % profile
                     print start_cmd + " ==> " + full_path
 
     def pre_construct(self):
         # IPClusterApp.pre_construct() is where we cd to the working directory.
         super(IPClusterApp, self).pre_construct()
         config = self.master_config
         try:
             daemon = config.Global.daemonize
             if daemon:
                 config.Global.log_to_file = True
         except AttributeError:
             pass
 
     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 start_launchers(self):
         config = self.master_config
 
         # Create the launchers. In both bases, we set the work_dir of
         # the launcher to the cluster_dir. This is where the launcher's
         # subprocesses will be launched. It is not where the controller
         # and engine will be launched.
         el_class = import_item(config.Global.engine_launcher)
         self.engine_launcher = el_class(
             work_dir=self.cluster_dir, config=config
         )
         cl_class = import_item(config.Global.controller_launcher)
         self.controller_launcher = cl_class(
             work_dir=self.cluster_dir, config=config
         )
 
         # Setup signals
         signal.signal(signal.SIGINT, self.sigint_handler)
 
         # Setup the observing of stopping. If the controller dies, shut
         # everything down as that will be completely fatal for the engines.
         d1 = self.controller_launcher.observe_stop()
         d1.addCallback(self.stop_launchers)
         # But, we don't monitor the stopping of engines. An engine dying
         # is just fine and in principle a user could start a new engine.
         # Also, if we did monitor engine stopping, it is difficult to
         # know what to do when only some engines die. Currently, the
         # observing of engine stopping is inconsistent. Some launchers
         # might trigger on a single engine stopping, other wait until
         # all stop.  TODO: think more about how to handle this.
         
         # Start the controller and engines
         self._stopping = False  # Make sure stop_launchers is not called 2x.
         d = self.start_controller()
         d.addCallback(self.start_engines)
         d.addCallback(self.startup_message)
         # If the controller or engines fail to start, stop everything
         d.addErrback(self.stop_launchers)
         return d
 
     def startup_message(self, r=None):
         log.msg("IPython cluster: started")
         return r
         
     def start_controller(self, r=None):
         # log.msg("In start_controller")
         config = self.master_config
         d = self.controller_launcher.start(
             cluster_dir=config.Global.cluster_dir
         )
         return d
             
     def start_engines(self, r=None):
         # log.msg("In start_engines")
         config = self.master_config
         d = self.engine_launcher.start(
             config.Global.n,
             cluster_dir=config.Global.cluster_dir
         )
         return d
 
     def stop_controller(self, r=None):
         # log.msg("In stop_controller")
         if self.controller_launcher.running:
             d = self.controller_launcher.stop()
             d.addErrback(self.log_err)
             return d
         else:
             return defer.succeed(None)
 
     def stop_engines(self, r=None):
         # log.msg("In stop_engines")
         if self.engine_launcher.running:
             d = self.engine_launcher.stop()
             d.addErrback(self.log_err)
             return d
         else:
             return defer.succeed(None)
 
     def log_err(self, f):
         log.msg(f.getTraceback())
         return None
         
     def stop_launchers(self, r=None):
         if not self._stopping:
             self._stopping = True
             if isinstance(r, failure.Failure):
                 log.msg('Unexpected error in ipcluster:')
                 log.msg(r.getTraceback())
             log.msg("IPython cluster: stopping")
             d= self.stop_engines()
             d2 = self.stop_controller()
             # Wait a few seconds to let things shut down.
-            reactor.callLater(3.0, reactor.stop)
+            reactor.callLater(4.0, reactor.stop)
 
     def sigint_handler(self, signum, frame):
         self.stop_launchers()
         
     def start_logging(self):
         # Remove old log files of the controller and engine
         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' + '-'):
                     if f.endswith('.log') or f.endswith('.out') or f.endswith('.err'):
                         os.remove(os.path.join(log_dir, f))
                 if f.startswith('ipcontroller' + '-'):
                     if f.endswith('.log') or f.endswith('.out') or f.endswith('.err'):
                         os.remove(os.path.join(log_dir, f))
         # This will remote old log files for ipcluster itself
         super(IPClusterApp, self).start_logging()
 
     def start_app(self):
         """Start the application, depending on what subcommand is used."""
         subcmd = self.master_config.Global.subcommand
         if subcmd=='create' or subcmd=='list':
             return
         elif subcmd=='start':
             self.start_app_start()
         elif subcmd=='stop':
             self.start_app_stop()
 
     def start_app_start(self):
         """Start the app for the start subcommand."""
         config = self.master_config
         # First see if the cluster is already running
         try:
             pid = self.get_pid_from_file()
         except PIDFileError:
             pass
         else:
             self.log.critical(
                 'Cluster is already running with [pid=%s]. '
                 'use "ipcluster stop" to stop the cluster.' % pid
             )
             # Here I exit with a unusual exit status that other processes
             # can watch for to learn how I existed.
             self.exit(ALREADY_STARTED)
 
         # Now log and daemonize
         self.log.info(
             'Starting ipcluster with [daemon=%r]' % config.Global.daemonize
         )
         if config.Global.daemonize:
             if os.name=='posix':
                 daemonize()
 
         # Now write the new pid file AFTER our new forked pid is active.
         self.write_pid_file()
         reactor.addSystemEventTrigger('during','shutdown', self.remove_pid_file)
         reactor.run()
 
     def start_app_stop(self):
         """Start the app for the stop subcommand."""
         config = self.master_config
         try:
             pid = self.get_pid_from_file()
         except PIDFileError:
             self.log.critical(
                 'Problem reading pid file, cluster is probably not running.'
             )
             # Here I exit with a unusual exit status that other processes
             # can watch for to learn how I existed.
             self.exit(ALREADY_STOPPED)
         else:
             if os.name=='posix':
                 sig = config.Global.signal
                 self.log.info(
                     "Stopping cluster [pid=%r] with [signal=%r]" % (pid, sig)
                 )
                 os.kill(pid, sig)
             elif os.name=='nt':
                 # As of right now, we don't support daemonize on Windows, so
                 # stop will not do anything. Minimally, it should clean up the
                 # old .pid files.
                 self.remove_pid_file()
 
 def launch_new_instance():
     """Create and run the IPython cluster."""
     app = IPClusterApp()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/kernel/multiengine.py

@@ -1,753 +1,752 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_multiengine -*-
 
 """Adapt the IPython ControllerServer to IMultiEngine.
 
 This module provides classes that adapt a ControllerService to the 
 IMultiEngine interface.  This interface is a basic interactive interface
 for working with a set of engines where it is desired to have explicit 
 access to each registered engine.  
 
 The classes here are exposed to the network in files like:
 
 * multienginevanilla.py
 * multienginepb.py
 """
 
 __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
 #-------------------------------------------------------------------------------
 
 from new import instancemethod
 from types import FunctionType
 
 from twisted.application import service
 from twisted.internet import defer, reactor
 from twisted.python import log, components, failure
 from zope.interface import Interface, implements, Attribute
 
 from IPython.utils import growl
 from IPython.kernel.util import printer
 from IPython.kernel.twistedutil import gatherBoth
 from IPython.kernel import map as Map
 from IPython.kernel import error
 from IPython.kernel.pendingdeferred import PendingDeferredManager, two_phase
 from IPython.kernel.controllerservice import \
     ControllerAdapterBase, \
     ControllerService, \
     IControllerBase
 
 
 #-------------------------------------------------------------------------------
 # Interfaces for the MultiEngine representation of a controller
 #-------------------------------------------------------------------------------
 
 class IEngineMultiplexer(Interface):
     """Interface to multiple engines implementing IEngineCore/Serialized/Queued.
     
     This class simply acts as a multiplexer of methods that are in the 
     various IEngines* interfaces.  Thus the methods here are jut like those 
     in the IEngine* interfaces, but with an extra first argument, targets.  
     The targets argument can have the following forms:
     
     * targets = 10            # Engines are indexed by ints
     * targets = [0,1,2,3]     # A list of ints
     * targets = 'all'         # A string to indicate all targets
     
     If targets is bad in any way, an InvalidEngineID will be raised.  This
     includes engines not being registered.
     
     All IEngineMultiplexer multiplexer methods must return a Deferred to a list 
     with length equal to the number of targets.  The elements of the list will 
     correspond to the return of the corresponding IEngine method.
     
     Failures are aggressive, meaning that if an action fails for any target, 
     the overall action will fail immediately with that Failure.
     
     :Parameters:
         targets : int, list of ints, or 'all'
             Engine ids the action will apply to.
     
     :Returns: Deferred to a list of results for each engine.
     
     :Exception:
         InvalidEngineID
             If the targets argument is bad or engines aren't registered.
         NoEnginesRegistered
             If there are no engines registered and targets='all'
     """
         
     #---------------------------------------------------------------------------
     # Mutiplexed methods
     #---------------------------------------------------------------------------
      
     def execute(lines, targets='all'):
         """Execute lines of Python code on targets.
               
         See the class docstring for information about targets and possible
         exceptions this method can raise.
               
         :Parameters:
             lines : str
                 String of python code to be executed on targets.
         """
             
     def push(namespace, targets='all'):
         """Push dict namespace into the user's namespace on targets.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             namspace : dict
                 Dict of key value pairs to be put into the users namspace.
         """
         
     def pull(keys, targets='all'):
         """Pull values out of the user's namespace on targets by keys.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             keys : tuple of strings
                 Sequence of keys to be pulled from user's namespace.
         """
           
     def push_function(namespace, targets='all'):
         """"""
         
     def pull_function(keys, targets='all'):
         """"""
                 
     def get_result(i=None, targets='all'):
         """Get the result for command i from targets.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             i : int or None
                 Command index or None to indicate most recent command.                
         """
         
     def reset(targets='all'):
         """Reset targets.
         
         This clears the users namespace of the Engines, but won't cause
         modules to be reloaded.
         """
         
     def keys(targets='all'):
         """Get variable names defined in user's namespace on targets."""
         
     def kill(controller=False, targets='all'):
         """Kill the targets Engines and possibly the controller.
         
         :Parameters:
             controller : boolean
                 Should the controller be killed as well.  If so all the 
                 engines will be killed first no matter what targets is.
         """
         
     def push_serialized(namespace, targets='all'):
         """Push a namespace of Serialized objects to targets.
         
         :Parameters:
             namespace : dict
                 A dict whose keys are the variable names and whose values
                 are serialized version of the objects.
         """
         
     def pull_serialized(keys, targets='all'):
         """Pull Serialized objects by keys from targets.
         
         :Parameters:
             keys : tuple of strings
                 Sequence of variable names to pull as serialized objects.
         """
         
     def clear_queue(targets='all'):
         """Clear the queue of pending command for targets."""
         
     def queue_status(targets='all'):
         """Get the status of the queue on the targets."""
     
     def set_properties(properties, targets='all'):
         """set properties by key and value"""
     
     def get_properties(keys=None, targets='all'):
         """get a list of properties by `keys`, if no keys specified, get all"""
     
     def del_properties(keys, targets='all'):
         """delete properties by `keys`"""
     
     def has_properties(keys, targets='all'):
         """get a list of bool values for whether `properties` has `keys`"""
     
     def clear_properties(targets='all'):
         """clear the properties dict"""
 
 
 class IMultiEngine(IEngineMultiplexer):
     """A controller that exposes an explicit interface to all of its engines.
     
     This is the primary inteface for interactive usage.
     """
     
     def get_ids():
         """Return list of currently registered ids.
                 
         :Returns:  A Deferred to a list of registered engine ids.
         """
 
 
 
 #-------------------------------------------------------------------------------
 # Implementation of the core MultiEngine classes
 #-------------------------------------------------------------------------------
 
 class MultiEngine(ControllerAdapterBase):
     """The representation of a ControllerService as a IMultiEngine.
     
     Although it is not implemented currently, this class would be where a
     client/notification API is implemented.  It could inherit from something
     like results.NotifierParent and then use the notify method to send
     notifications.
     """
     
     implements(IMultiEngine)
     
     def __init(self, controller):
         ControllerAdapterBase.__init__(self, controller)
     
     #---------------------------------------------------------------------------
     # Helper methods
     #---------------------------------------------------------------------------
     
     def engineList(self, targets):
         """Parse the targets argument into a list of valid engine objects.
         
         :Parameters:
             targets : int, list of ints or 'all'
                 The targets argument to be parsed.
                 
         :Returns: List of engine objects.
         
         :Exception:
             InvalidEngineID
                 If targets is not valid or if an engine is not registered.
         """
         if isinstance(targets, int):
             if targets not in self.engines.keys():
                 log.msg("Engine with id %i is not registered" % targets)
                 raise error.InvalidEngineID("Engine with id %i is not registered" % targets)
             else: 
                 return [self.engines[targets]]
         elif isinstance(targets, (list, tuple)):
             for id in targets:
                 if id not in self.engines.keys():
                     log.msg("Engine with id %r is not registered" % id)
                     raise error.InvalidEngineID("Engine with id %r is not registered" % id)  
             return map(self.engines.get, targets)
         elif targets == 'all':
             eList = self.engines.values()
             if len(eList) == 0:
-                msg = """There are no engines registered.
-                     Check the logs in ~/.ipython/log if you think there should have been."""
-                raise error.NoEnginesRegistered(msg)
+                raise error.NoEnginesRegistered("There are no engines registered. "
+                    "Check the logs if you think there should have been.")
             else:
                 return eList
         else:
             raise error.InvalidEngineID("targets argument is not an int, list of ints or 'all': %r"%targets)
     
     def _performOnEngines(self, methodName, *args, **kwargs):
         """Calls a method on engines and returns deferred to list of results.
         
         :Parameters:
             methodName : str
                 Name of the method to be called.
             targets : int, list of ints, 'all'
                 The targets argument to be parsed into a list of engine objects.
             args
                 The positional keyword arguments to be passed to the engines.
             kwargs
                 The keyword arguments passed to the method
                 
         :Returns: List of deferreds to the results on each engine
         
         :Exception:
             InvalidEngineID
                 If the targets argument is bad in any way.
             AttributeError
                 If the method doesn't exist on one of the engines.
         """
         targets = kwargs.pop('targets')
         log.msg("Performing %s on %r" % (methodName, targets))
         # log.msg("Performing %s(%r, %r) on %r" % (methodName, args, kwargs, targets))
         # This will and should raise if targets is not valid!
         engines = self.engineList(targets)
         dList = []
         for e in engines:
             meth = getattr(e, methodName, None)
             if meth is not None:
                 dList.append(meth(*args, **kwargs))
             else:
                 raise AttributeError("Engine %i does not have method %s" % (e.id, methodName))
         return dList
         
     def _performOnEnginesAndGatherBoth(self, methodName, *args, **kwargs):
         """Called _performOnEngines and wraps result/exception into deferred."""
         try:
             dList = self._performOnEngines(methodName, *args, **kwargs)
         except (error.InvalidEngineID, AttributeError, KeyError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())
         else:
             # Having fireOnOneErrback is causing problems with the determinacy
             # of the system.  Basically, once a single engine has errbacked, this
             # method returns.  In some cases, this will cause client to submit
             # another command.  Because the previous command is still running
             # on some engines, this command will be queued.  When those commands
             # then errback, the second command will raise QueueCleared.  Ahhh!
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, methodName)
             return d
               
     #---------------------------------------------------------------------------
     # General IMultiEngine methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         return defer.succeed(self.engines.keys())
     
     #---------------------------------------------------------------------------
     # IEngineMultiplexer methods
     #---------------------------------------------------------------------------
     
     def execute(self, lines, targets='all'):
         return self._performOnEnginesAndGatherBoth('execute', lines, targets=targets)
     
     def push(self, ns, targets='all'):
         return self._performOnEnginesAndGatherBoth('push', ns, targets=targets)
         
     def pull(self, keys, targets='all'):
         return self._performOnEnginesAndGatherBoth('pull', keys, targets=targets)
     
     def push_function(self, ns, targets='all'):
         return self._performOnEnginesAndGatherBoth('push_function', ns, targets=targets)
         
     def pull_function(self, keys, targets='all'):
         return self._performOnEnginesAndGatherBoth('pull_function', keys, targets=targets)
     
     def get_result(self, i=None, targets='all'):
         return self._performOnEnginesAndGatherBoth('get_result', i, targets=targets)
     
     def reset(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('reset', targets=targets)
     
     def keys(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('keys', targets=targets)
     
     def kill(self, controller=False, targets='all'):
         if controller:
             targets = 'all'
         d = self._performOnEnginesAndGatherBoth('kill', targets=targets)
         if controller:
             log.msg("Killing controller")
             d.addCallback(lambda _: reactor.callLater(2.0, reactor.stop))
             # Consume any weird stuff coming back
             d.addBoth(lambda _: None)
         return d
     
     def push_serialized(self, namespace, targets='all'):
         for k, v in namespace.iteritems():
             log.msg("Pushed object %s is %f MB" % (k, v.getDataSize()))
         d = self._performOnEnginesAndGatherBoth('push_serialized', namespace, targets=targets)      
         return d
         
     def pull_serialized(self, keys, targets='all'):
         try:
             dList = self._performOnEngines('pull_serialized', keys, targets=targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())
         else:
             for d in dList:
                 d.addCallback(self._logSizes)
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'pull_serialized')
             return d  
                               
     def _logSizes(self, listOfSerialized):
         if isinstance(listOfSerialized, (list, tuple)):
             for s in listOfSerialized:
                 log.msg("Pulled object is %f MB" % s.getDataSize())
         else:
             log.msg("Pulled object is %f MB" % listOfSerialized.getDataSize())
         return listOfSerialized
     
     def clear_queue(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('clear_queue', targets=targets)         
     
     def queue_status(self, targets='all'):
         log.msg("Getting queue status on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = []
             for e in engines:
                 dList.append(e.queue_status().addCallback(lambda s:(e.id, s)))
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'queue_status')
             return d 
     
     def get_properties(self, keys=None, targets='all'):
         log.msg("Getting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.get_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'get_properties')
             return d
     
     def set_properties(self, properties, targets='all'):
         log.msg("Setting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.set_properties(properties) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'set_properties')
             return d
     
     def has_properties(self, keys, targets='all'):
         log.msg("Checking properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.has_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'has_properties')
             return d
     
     def del_properties(self, keys, targets='all'):
         log.msg("Deleting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.del_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'del_properties')
             return d
     
     def clear_properties(self, targets='all'):
         log.msg("Clearing properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.clear_properties() for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'clear_properties')
             return d
 
 
 components.registerAdapter(MultiEngine,
                            IControllerBase, 
                            IMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # Interfaces for the Synchronous MultiEngine
 #-------------------------------------------------------------------------------
 
 class ISynchronousEngineMultiplexer(Interface):
     pass
 
 
 class ISynchronousMultiEngine(ISynchronousEngineMultiplexer):
     """Synchronous, two-phase version of IMultiEngine.
     
     Methods in this interface are identical to those of IMultiEngine, but they
     take one additional argument:
     
     execute(lines, targets='all') -> execute(lines, targets='all, block=True)
     
     :Parameters:
         block : boolean
             Should the method return a deferred to a deferredID or the 
             actual result.  If block=False a deferred to a deferredID is 
             returned and the user must call `get_pending_deferred` at a later
             point.  If block=True, a deferred to the actual result comes back.
     """
     def get_pending_deferred(deferredID, block=True):
         """"""
     
     def clear_pending_deferreds():
         """"""
 
 
 #-------------------------------------------------------------------------------
 # Implementation of the Synchronous MultiEngine
 #-------------------------------------------------------------------------------
 
 class SynchronousMultiEngine(PendingDeferredManager):
     """Adapt an `IMultiEngine` -> `ISynchronousMultiEngine`
     
     Warning, this class uses a decorator that currently uses **kwargs.  
     Because of this block must be passed as a kwarg, not positionally.
     """
     
     implements(ISynchronousMultiEngine)
     
     def __init__(self, multiengine):
         self.multiengine = multiengine
         PendingDeferredManager.__init__(self)
     
     #---------------------------------------------------------------------------
     # Decorated pending deferred methods
     #---------------------------------------------------------------------------
     
     @two_phase
     def execute(self, lines, targets='all'):
         d = self.multiengine.execute(lines, targets)
         return d
     
     @two_phase
     def push(self, namespace, targets='all'):
         return self.multiengine.push(namespace, targets)
     
     @two_phase
     def pull(self, keys, targets='all'):
         d = self.multiengine.pull(keys, targets)
         return d
     
     @two_phase
     def push_function(self, namespace, targets='all'):
         return self.multiengine.push_function(namespace, targets)
     
     @two_phase
     def pull_function(self, keys, targets='all'):
         d = self.multiengine.pull_function(keys, targets)
         return d
     
     @two_phase
     def get_result(self, i=None, targets='all'):
         return self.multiengine.get_result(i, targets='all')
     
     @two_phase
     def reset(self, targets='all'):
         return self.multiengine.reset(targets)
     
     @two_phase
     def keys(self, targets='all'):
         return self.multiengine.keys(targets)
     
     @two_phase
     def kill(self, controller=False, targets='all'):
         return self.multiengine.kill(controller, targets)
     
     @two_phase
     def push_serialized(self, namespace, targets='all'):
         return self.multiengine.push_serialized(namespace, targets)
     
     @two_phase
     def pull_serialized(self, keys, targets='all'):
         return self.multiengine.pull_serialized(keys, targets)
     
     @two_phase
     def clear_queue(self, targets='all'):
         return self.multiengine.clear_queue(targets)
     
     @two_phase
     def queue_status(self, targets='all'):
         return self.multiengine.queue_status(targets)
     
     @two_phase
     def set_properties(self, properties, targets='all'):
         return self.multiengine.set_properties(properties, targets)
     
     @two_phase
     def get_properties(self, keys=None, targets='all'):
         return self.multiengine.get_properties(keys, targets)
     
     @two_phase
     def has_properties(self, keys, targets='all'):
         return self.multiengine.has_properties(keys, targets)
     
     @two_phase
     def del_properties(self, keys, targets='all'):
         return self.multiengine.del_properties(keys, targets)
     
     @two_phase
     def clear_properties(self, targets='all'):
         return self.multiengine.clear_properties(targets)
     
     #---------------------------------------------------------------------------
     # IMultiEngine methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         """Return a list of registered engine ids.
         
         Never use the two phase block/non-block stuff for this.
         """
         return self.multiengine.get_ids()
 
 
 components.registerAdapter(SynchronousMultiEngine, IMultiEngine, ISynchronousMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # Various high-level interfaces that can be used as MultiEngine mix-ins
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # IMultiEngineCoordinator
 #-------------------------------------------------------------------------------
 
 class IMultiEngineCoordinator(Interface):
     """Methods that work on multiple engines explicitly."""
        
     def scatter(key, seq, dist='b', flatten=False, targets='all'):
         """Partition and distribute a sequence to targets."""
         
     def gather(key, dist='b', targets='all'):
         """Gather object key from targets."""
     
     def raw_map(func, seqs, dist='b', targets='all'):
         """
         A parallelized version of Python's builtin `map` function.
         
         This has a slightly different syntax than the builtin `map`.
         This is needed because we need to have keyword arguments and thus
         can't use *args to capture all the sequences.  Instead, they must
         be passed in a list or tuple.
         
         The equivalence is:
         
         raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
         
         Most users will want to use parallel functions or the `mapper`
         and `map` methods for an API that follows that of the builtin
         `map`.
         """
 
 
 class ISynchronousMultiEngineCoordinator(IMultiEngineCoordinator):
     """Methods that work on multiple engines explicitly."""
 
     def scatter(key, seq, dist='b', flatten=False, targets='all', block=True):
         """Partition and distribute a sequence to targets."""
         
     def gather(key, dist='b', targets='all', block=True):
         """Gather object key from targets"""
     
     def raw_map(func, seqs, dist='b', targets='all', block=True):
         """
         A parallelized version of Python's builtin map.
         
         This has a slightly different syntax than the builtin `map`.
         This is needed because we need to have keyword arguments and thus
         can't use *args to capture all the sequences.  Instead, they must
         be passed in a list or tuple.
         
         raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
         
         Most users will want to use parallel functions or the `mapper`
         and `map` methods for an API that follows that of the builtin
         `map`.
         """
 
 
 #-------------------------------------------------------------------------------
 # IMultiEngineExtras
 #-------------------------------------------------------------------------------
 
 class IMultiEngineExtras(Interface):
     
     def zip_pull(targets, keys):
         """
         Pull, but return results in a different format from `pull`.
         
         This method basically returns zip(pull(targets, *keys)), with a few 
         edge cases handled differently.  Users of chainsaw will find this format 
         familiar.
         """
     
     def run(targets, fname):
         """Run a .py file on targets."""
 
 
 class ISynchronousMultiEngineExtras(IMultiEngineExtras):
     def zip_pull(targets, keys, block=True):
         """
         Pull, but return results in a different format from `pull`.
         
         This method basically returns zip(pull(targets, *keys)), with a few 
         edge cases handled differently.  Users of chainsaw will find this format 
         familiar.
         """
     
     def run(targets, fname, block=True):
         """Run a .py file on targets."""
 
 #-------------------------------------------------------------------------------
 # The full MultiEngine interface
 #-------------------------------------------------------------------------------
 
 class IFullMultiEngine(IMultiEngine, 
     IMultiEngineCoordinator, 
     IMultiEngineExtras):
     pass
 
 
 class IFullSynchronousMultiEngine(ISynchronousMultiEngine, 
     ISynchronousMultiEngineCoordinator,
     ISynchronousMultiEngineExtras):
     pass