Fixed bugs in IPython.kernel.

* Removed :mod:`IPython.kernel.contexts` and
  :mod:`IPython.kernel.tests.test_contexts` and all code that used them. This
  was broken and needed to be removed. Also Enthought has a nice decorator
  that does this. * Fixed broken tests in :file:`test_multienginefc.py` and
  :file:`test_taskfc.py`. See https://bugs.launchpad.net/ipython/+bug/505358.
* Changed ports number in :mod:`IPython.kernel.tests` to 10111 to avoid
  conflicts with users who are using the older 10105 ports. See
  https://bugs.launchpad.net/ipython/+bug/504515.
* Fixed logic and bugs in :mod:`IPython.kernel.fcutils` and
  :mod:`IPython.kernel.clientconnector` related to how FURL files are 
  validated and found. See https://bugs.launchpad.net/ipython/+bug/502790.
* Fixed the unhandled :class:`DeadReferenceError`` exception that was happening
  in :mod:`IPython.kernel.multiengineclient' and :mod:`IPython.kernel.taskclient`
  when the controller died or restarted. Now we capture that error
  and raises a :class:`IPython.kernel.error.ConnectionError` that has a
  better error message. See https://bugs.launchpad.net/ipython/+bug/502796.

IPython/kernel/clientconnector.py

@@ -1,773 +1,788 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,
+    is_valid_furl_file,
     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.path 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.
+        """Find a FURL file.
+
+        If successful, this returns a FURL file that exists on the file
+        system. The contents of the file have not been checked though. This
+        is because we often have to deal with FURL file whose buffers have
+        not been flushed.
 
         This raises an :exc:`~IPython.kernel.fcutil.FURLError` exception 
         if a FURL file can't be found.
+
+        This tries the following:
+        
+        1. By the name ``furl_or_file``.
+        2. By ``cluster_dir`` and ``furl_file_name``.
+        3. By cluster profile with a default of ``default``. This uses
+           ``ipython_dir``.
         """
         # Try by furl_or_file
         if furl_or_file is not None:
-            validate_furl_or_file(furl_or_file)
-            return furl_or_file
+            if is_valid_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')
+            raise FURLError('A furl_file_name must be provided if furl_or_file is not')
 
         # 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 
+            A furl or a filename containing a FURL. 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 
+            A furl or a filename containing a FURL. 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
             )
+            # If this succeeds, we know the furl file exists and has a .furl
+            # extension, but it could still be empty. That is checked each
+            # connection attempt.
         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 [%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 
+            A furl or a filename containing a FURL. 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 
+            A furl or a filename containing a FURL. 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/fcutil.py

@@ -1,277 +1,293 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 Foolscap related utilities.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from __future__ import with_statement
 
 import os
 import tempfile
 
 from twisted.internet import reactor, defer
 from twisted.python import log
 
+import foolscap
 from foolscap import Tub, UnauthenticatedTub
 
 from IPython.config.loader import Config
-
 from IPython.kernel.configobjfactory import AdaptedConfiguredObjectFactory
-
 from IPython.kernel.error import SecurityError
 
-from IPython.utils.traitlets import Int, Str, Bool, Instance
 from IPython.utils.importstring import import_item
+from IPython.utils.path import expand_path
+from IPython.utils.traitlets import Int, Str, Bool, Instance
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 
 # We do this so if a user doesn't have OpenSSL installed, it will try to use
 # an UnauthenticatedTub.  But, they will still run into problems if they
 # try to use encrypted furls.
 try:
     import OpenSSL
 except:
     Tub = UnauthenticatedTub
     have_crypto = False
 else:
     have_crypto = True
 
 
 class FURLError(Exception):
     pass
 
 
 def check_furl_file_security(furl_file, secure):
     """Remove the old furl_file if changing security modes."""
+    furl_file = expand_path(furl_file)
     if os.path.isfile(furl_file):
-        f = open(furl_file, 'r')
-        oldfurl = f.read().strip()
-        f.close()
+        with open(furl_file, 'r') as f:
+            oldfurl = f.read().strip()
         if (oldfurl.startswith('pb://') and not secure) or (oldfurl.startswith('pbu://') and secure):
             os.remove(furl_file)
 
 
 def is_secure(furl):
     """Is the given FURL secure or not."""
-    if is_valid(furl):
+    if is_valid_furl(furl):
         if furl.startswith("pb://"):
             return True
         elif furl.startswith("pbu://"):
             return False
     else:
         raise FURLError("invalid FURL: %s" % furl)
 
 
-def is_valid(furl):
+def is_valid_furl(furl):
     """Is the str a valid FURL or not."""
     if isinstance(furl, str):
         if furl.startswith("pb://") or furl.startswith("pbu://"):
             return True
+        else:
+            return False
     else:
         return False
 
 
+def is_valid_furl_file(furl_or_file):
+    """See if furl_or_file exists and contains a valid FURL.
+
+    This doesn't try to read the contents because often we have to validate
+    FURL files that are created, but don't yet have a FURL written to them.
+    """
+    if isinstance(furl_or_file, (str, unicode)):
+        path, furl_filename = os.path.split(furl_or_file)
+        if os.path.isdir(path) and furl_filename.endswith('.furl'):
+            return True
+    return False
+
+
 def find_furl(furl_or_file):
-    """Find, validate and return a FURL in a string or file."""
-    if isinstance(furl_or_file, str):
-        if is_valid(furl_or_file):
-            return furl_or_file
-    if os.path.isfile(furl_or_file):
+    """Find, validate and return a FURL in a string or file.
+
+    This calls :func:`IPython.utils.path.expand_path` on the argument to
+    properly handle ``~`` and ``$`` variables in the path.
+    """
+    if is_valid_furl(furl_or_file):
+        return furl_or_file
+    furl_or_file = expand_path(furl_or_file)
+    if is_valid_furl_file(furl_or_file):
         with open(furl_or_file, 'r') as f:
             furl = f.read().strip()
-        if is_valid(furl):
+        if is_valid_furl(furl):
             return furl
-    raise FURLError("Not a valid FURL or FURL file: %s" % furl_or_file)
+    raise FURLError("Not a valid FURL or FURL file: %r" % furl_or_file)
 
 
 def is_valid_furl_or_file(furl_or_file):
     """Validate a FURL or a FURL file.
 
     If ``furl_or_file`` looks like a file, we simply make sure its directory
     exists and that it has a ``.furl`` file extension.  We don't try to see
     if the FURL file exists or to read its contents. This is useful for
     cases where auto re-connection is being used.
     """
-    if isinstance(furl_or_file, str):
-        if is_valid(furl_or_file):
-            return True
-    if isinstance(furl_or_file, (str, unicode)):
-        path, furl_filename = os.path.split(furl_or_file)
-        if os.path.isdir(path) and furl_filename.endswith('.furl'):
-            return True
-    return False
+    if is_valid_furl(furl_or_file) or is_valid_furl_file(furl_or_file):
+        return True
+    else:
+        return False
 
 
 def validate_furl_or_file(furl_or_file):
+    """Like :func:`is_valid_furl_or_file`, but raises an error."""
     if not is_valid_furl_or_file(furl_or_file):
         raise FURLError('Not a valid FURL or FURL file: %r' % furl_or_file)
 
 
 def get_temp_furlfile(filename):
     """Return a temporary FURL file."""
     return tempfile.mktemp(dir=os.path.dirname(filename),
                            prefix=os.path.basename(filename))
 
 
 def make_tub(ip, port, secure, cert_file):
     """Create a listening tub given an ip, port, and cert_file location.
     
     Parameters
     ----------
     ip : str
         The ip address or hostname that the tub should listen on.  
         Empty means all interfaces.
     port : int
         The port that the tub should listen on.  A value of 0 means
         pick a random port
     secure: bool
         Will the connection be secure (in the Foolscap sense).
     cert_file: str
         A filename of a file to be used for theSSL certificate.
 
     Returns
     -------
     A tub, listener tuple.
     """
     if secure:
         if have_crypto:
             tub = Tub(certFile=cert_file)
         else:
             raise SecurityError("OpenSSL/pyOpenSSL is not available, so we "
                                 "can't run in secure mode.  Try running without "
                                 "security using 'ipcontroller -xy'.")
     else:
         tub = UnauthenticatedTub()
     
     # Set the strport based on the ip and port and start listening
     if ip == '':
         strport = "tcp:%i" % port
     else:
         strport = "tcp:%i:interface=%s" % (port, ip)
     log.msg("Starting listener with [secure=%r] on: %s" % (secure, strport))
     listener = tub.listenOn(strport)
     
     return tub, listener
 
 
 class FCServiceFactory(AdaptedConfiguredObjectFactory):
     """This class creates a tub with various services running in it.
 
     The basic idea is that :meth:`create` returns a running :class:`Tub`
     instance that has a number of Foolscap references registered in it.
     This class is a subclass of :class:`IPython.core.component.Component`
     so the IPython configuration and component system are used.
 
     Attributes
     ----------
     interfaces : Config
         A Config instance whose values are sub-Config objects having two
         keys: furl_file and interface_chain.
 
     The other attributes are the standard ones for Foolscap.
     """
 
     ip = Str('', config=True)
     port = Int(0, config=True)
     secure = Bool(True, config=True)
     cert_file = Str('', config=True)
     location = Str('', config=True)
     reuse_furls = Bool(False, config=True)
     interfaces = Instance(klass=Config, kw={}, allow_none=False, config=True)
 
     def __init__(self, config, adaptee):
         super(FCServiceFactory, self).__init__(config, adaptee)
         self._check_reuse_furls()
 
     def _ip_changed(self, name, old, new):
         if new == 'localhost' or new == '127.0.0.1':
             self.location = '127.0.0.1'
 
     def _check_reuse_furls(self):
         furl_files = [i.furl_file for i in self.interfaces.values()]
         for ff in furl_files:
             fullfile = self._get_security_file(ff)
             if self.reuse_furls:
                 if self.port==0:
                     raise FURLError("You are trying to reuse the FURL file "
                         "for this connection, but the port for this connection "
                         "is set to 0 (autoselect). To reuse the FURL file "
                         "you need to specify specific port to listen on."
                     )
                 else:
                     log.msg("Reusing FURL file: %s" % fullfile)
             else:
                 if os.path.isfile(fullfile):
                     log.msg("Removing old FURL file: %s" % fullfile)
                     os.remove(fullfile)
 
     def _get_security_file(self, filename):
         return os.path.join(self.config.Global.security_dir, filename)
 
     def create(self):
         """Create and return the Foolscap tub with everything running."""
 
         self.tub, self.listener = make_tub(
             self.ip, self.port, self.secure, 
             self._get_security_file(self.cert_file)
         )
         # log.msg("Interfaces to register [%r]: %r" % \
         #     (self.__class__, self.interfaces))
         if not self.secure:
             log.msg("WARNING: running with no security: %s" % \
                 self.__class__.__name__)
         reactor.callWhenRunning(self.set_location_and_register)
         return self.tub
 
     def set_location_and_register(self):
         """Set the location for the tub and return a deferred."""
 
         if self.location == '':
             d = self.tub.setLocationAutomatically()
         else:
             d = defer.maybeDeferred(self.tub.setLocation,
                 "%s:%i" % (self.location, self.listener.getPortnum()))
         self.adapt_to_interfaces(d)
 
     def adapt_to_interfaces(self, d):
         """Run through the interfaces, adapt and register."""
 
         for ifname, ifconfig in self.interfaces.iteritems():
             ff = self._get_security_file(ifconfig.furl_file)
             log.msg("Adapting [%s] to interface: %s" % \
                 (self.adaptee.__class__.__name__, ifname))
             log.msg("Saving FURL for interface [%s] to file: %s" % (ifname, ff))
             check_furl_file_security(ff, self.secure)
             adaptee = self.adaptee
             for i in ifconfig.interface_chain:
                 adaptee = import_item(i)(adaptee)
             d.addCallback(self.register, adaptee, furl_file=ff)
 
     def register(self, empty, ref, furl_file):
         """Register the reference with the FURL file.
 
         The FURL file is created and then moved to make sure that when the
         file appears, the buffer has been flushed and the file closed. This
         is not done if we are re-using FURLS however.
         """
         if self.reuse_furls:
             self.tub.registerReference(ref, furlFile=furl_file)
         else:
             temp_furl_file = get_temp_furlfile(furl_file)
             self.tub.registerReference(ref, furlFile=temp_furl_file)
             os.rename(temp_furl_file, furl_file)
 

IPython/kernel/multiengineclient.py

@@ -1,967 +1,901 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_multiengineclient -*-
 
 """General Classes for IMultiEngine clients."""
 
 __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 sys
-import linecache
 import warnings
 
 from twisted.python import components
 from twisted.python.failure import Failure
 from zope.interface import Interface, implements, Attribute
+from foolscap import DeadReferenceError
 
 from IPython.utils.coloransi import TermColors
 
 from IPython.kernel.twistedutil import blockingCallFromThread
 from IPython.kernel import error
 from IPython.kernel.parallelfunction import ParallelFunction
 from IPython.kernel.mapper import (
     MultiEngineMapper, 
     IMultiEngineMapperFactory,
     IMapper
 )
 
 from IPython.kernel.multiengine import IFullSynchronousMultiEngine
 
 
 #-------------------------------------------------------------------------------
 # Pending Result things
 #-------------------------------------------------------------------------------
 
 class IPendingResult(Interface):
     """A representation of a result that is pending.
     
     This class is similar to Twisted's `Deferred` object, but is designed to be
     used in a synchronous context.
     """
     
     result_id=Attribute("ID of the deferred on the other side")
     client=Attribute("A client that I came from")
     r=Attribute("An attribute that is a property that calls and returns get_result")
     
     def get_result(default=None, block=True):
         """
         Get a result that is pending.
                 
         :Parameters:
             default
                 The value to return if the result is not ready.
             block : boolean
                 Should I block for the result.
                 
         :Returns: The actual result or the default value.
         """
         
     def add_callback(f, *args, **kwargs):
         """
         Add a callback that is called with the result.
         
         If the original result is foo, adding a callback will cause
         f(foo, *args, **kwargs) to be returned instead.  If multiple
         callbacks are registered, they are chained together: the result of
         one is passed to the next and so on.  
         
         Unlike Twisted's Deferred object, there is no errback chain.  Thus
         any exception raised will not be caught and handled.  User must 
         catch these by hand when calling `get_result`.
         """
 
 
 class PendingResult(object):
     """A representation of a result that is not yet ready.
     
     A user should not create a `PendingResult` instance by hand.
     
     Methods:
     
     * `get_result`
     * `add_callback`
     
     Properties:
     
     * `r`
     """
     
     def __init__(self, client, result_id):
         """Create a PendingResult with a result_id and a client instance.
         
         The client should implement `_getPendingResult(result_id, block)`.
         """
         self.client = client
         self.result_id = result_id
         self.called = False
         self.raised = False
         self.callbacks = []
         
     def get_result(self, default=None, block=True):
         """Get a result that is pending.
                 
         This method will connect to an IMultiEngine adapted controller
         and see if the result is ready.  If the action triggers an exception
         raise it and record it.  This method records the result/exception once it is 
         retrieved.  Calling `get_result` again will get this cached result or will
         re-raise the exception.  The .r attribute is a property that calls
         `get_result` with block=True.
         
         :Parameters:
             default
                 The value to return if the result is not ready.
             block : boolean
                 Should I block for the result.
                 
         :Returns: The actual result or the default value.
         """
         
         if self.called:
             if self.raised:
                 raise self.result[0], self.result[1], self.result[2]
             else:
                 return self.result
         try:
             result = self.client.get_pending_deferred(self.result_id, block)
         except error.ResultNotCompleted:
             return default
         except:
             # Reraise other error, but first record them so they can be reraised
             # later if .r or get_result is called again.
             self.result = sys.exc_info()
             self.called = True
             self.raised = True
             raise
         else:
             for cb in self.callbacks:
                 result = cb[0](result, *cb[1], **cb[2])
             self.result = result
             self.called = True
             return result
         
     def add_callback(self, f, *args, **kwargs):
         """Add a callback that is called with the result.
         
         If the original result is result, adding a callback will cause
         f(result, *args, **kwargs) to be returned instead.  If multiple
         callbacks are registered, they are chained together: the result of
         one is passed to the next and so on.  
         
         Unlike Twisted's Deferred object, there is no errback chain.  Thus
         any exception raised will not be caught and handled.  User must 
         catch these by hand when calling `get_result`.
         """
         assert callable(f)
         self.callbacks.append((f, args, kwargs))
         
     def __cmp__(self, other):
         if self.result_id < other.result_id:
             return -1
         else:
             return 1
             
     def _get_r(self):
         return self.get_result(block=True)
     
     r = property(_get_r)
     """This property is a shortcut to a `get_result(block=True)`."""
 
 
 #-------------------------------------------------------------------------------
 # Pretty printing wrappers for certain lists
 #-------------------------------------------------------------------------------    
     
 class ResultList(list):
     """A subclass of list that pretty prints the output of `execute`/`get_result`."""
     
     def __repr__(self):
         output = []
         # These colored prompts were not working on Windows
         if sys.platform == 'win32':
             blue = normal = red = green = ''
         else:
             blue = TermColors.Blue
             normal = TermColors.Normal
             red = TermColors.Red
             green = TermColors.Green
         output.append("<Results List>\n")
         for cmd in self:
             if isinstance(cmd, Failure):
                 output.append(cmd)
             else:
                 target = cmd.get('id',None)
                 cmd_num = cmd.get('number',None)
                 cmd_stdin = cmd.get('input',{}).get('translated','No Input')
                 cmd_stdout = cmd.get('stdout', None)
                 cmd_stderr = cmd.get('stderr', None)
                 output.append("%s[%i]%s In [%i]:%s %s\n" % \
                     (green, target,
                     blue, cmd_num, normal, cmd_stdin))
                 if cmd_stdout:
                     output.append("%s[%i]%s Out[%i]:%s %s\n" % \
                         (green, target,
                         red, cmd_num, normal, cmd_stdout))
                 if cmd_stderr:
                     output.append("%s[%i]%s Err[%i]:\n%s %s" % \
                         (green, target,
                         red, cmd_num, normal, cmd_stderr))
         return ''.join(output)
 
 
 def wrapResultList(result):
     """A function that wraps the output of `execute`/`get_result` -> `ResultList`."""
     if len(result) == 0:
         result = [result]
     return ResultList(result)
 
 
 class QueueStatusList(list):
     """A subclass of list that pretty prints the output of `queue_status`."""
     
     def __repr__(self):
         output = []
         output.append("<Queue Status List>\n")
         for e in self:
             output.append("Engine: %s\n" % repr(e[0]))
             output.append("    Pending: %s\n" % repr(e[1]['pending']))
             for q in e[1]['queue']:
                 output.append("    Command: %s\n" % repr(q))
         return ''.join(output)
 
 
 #-------------------------------------------------------------------------------
 # InteractiveMultiEngineClient
 #-------------------------------------------------------------------------------    
 
 class InteractiveMultiEngineClient(object):
     """A mixin class that add a few methods to a multiengine client.
     
     The methods in this mixin class are designed for interactive usage.
     """
                                         
     def activate(self):
         """Make this `MultiEngineClient` active for parallel magic commands.
         
         IPython has a magic command syntax to work with `MultiEngineClient` objects.
         In a given IPython session there is a single active one.  While
         there can be many `MultiEngineClient` created and used by the user, 
         there is only one active one.  The active `MultiEngineClient` is used whenever 
         the magic commands %px and %autopx are used.
         
         The activate() method is called on a given `MultiEngineClient` to make it 
         active.  Once this has been done, the magic commands can be used.
         """
         
         try:
             # This is injected into __builtins__.
             ip = get_ipython()
         except NameError:
             print "The IPython parallel magics (%result, %px, %autopx) only work within IPython."
         else:
             pmagic = ip.get_component('parallel_magic')
             if pmagic is not None:
                 pmagic.active_multiengine_client = self
             else:
                 print "You must first load the parallelmagic extension " \
                       "by doing '%load_ext parallelmagic'"
 
     def __setitem__(self, key, value):
         """Add a dictionary interface for pushing/pulling.
         
         This functions as a shorthand for `push`.
         
         :Parameters:
             key : str 
                 What to call the remote object.
             value : object
                 The local Python object to push.
         """
         targets, block = self._findTargetsAndBlock()
         return self.push({key:value}, targets=targets, block=block)
     
     def __getitem__(self, key):
         """Add a dictionary interface for pushing/pulling.
         
         This functions as a shorthand to `pull`.
         
         :Parameters:
          - `key`: A string representing the key.
         """
         if isinstance(key, str):
             targets, block = self._findTargetsAndBlock()
             return self.pull(key, targets=targets, block=block)
         else:
             raise TypeError("__getitem__ only takes strs")
             
     def __len__(self):
         """Return the number of available engines."""
         return len(self.get_ids())
-    
-    #---------------------------------------------------------------------------
-    # Make this a context manager for with
-    #---------------------------------------------------------------------------
-    
-    def findsource_file(self,f):
-        linecache.checkcache()
-        s = findsource(f.f_code) # findsource is not defined!
-        lnum = f.f_lineno
-        wsource = s[0][f.f_lineno:]
-        return strip_whitespace(wsource)
-
-    def findsource_ipython(self,f):
-        from IPython.core import ipapi
-        self.ip = ipapi.get()
-        wsource = [l+'\n' for l in
-                   self.ip.input_hist_raw[-1].splitlines()[1:]] 
-        return strip_whitespace(wsource)
-        
-    def __enter__(self):
-        f = sys._getframe(1)
-        local_ns = f.f_locals
-        global_ns = f.f_globals
-        if f.f_code.co_filename == '<ipython console>':
-            s = self.findsource_ipython(f)
-        else:
-            s = self.findsource_file(f)
-
-        self._with_context_result = self.execute(s)
-
-    def __exit__ (self, etype, value, tb):
-        if issubclass(etype,error.StopLocalExecution):
-            return True
-
-
-def remote():
-    m = 'Special exception to stop local execution of parallel code.'
-    raise error.StopLocalExecution(m)
-
-def strip_whitespace(source):
-    # Expand tabs to avoid any confusion.
-    wsource = [l.expandtabs(4) for l in source]
-    # Detect the indentation level
-    done = False
-    for line in wsource:
-        if line.isspace():
-            continue
-        for col,char in enumerate(line):
-            if char != ' ':
-                done = True
-                break
-        if done:
-            break
-    # Now we know how much leading space there is in the code.  Next, we
-    # extract up to the first line that has less indentation.
-    # WARNINGS: we skip comments that may be misindented, but we do NOT yet
-    # detect triple quoted strings that may have flush left text.
-    for lno,line in enumerate(wsource):
-        lead = line[:col]
-        if lead.isspace():
-            continue
-        else:
-            if not lead.lstrip().startswith('#'):
-                break
-    # The real 'with' source is up to lno
-    src_lines = [l[col:] for l in wsource[:lno+1]]
-
-    # Finally, check that the source's first non-comment line begins with the
-    # special call 'remote()'
-    for nline,line in enumerate(src_lines):
-        if line.isspace() or line.startswith('#'):
-            continue
-        if 'remote()' in line:
-            break
-        else:
-            raise ValueError('remote() call missing at the start of code')
-    src = ''.join(src_lines[nline+1:])
-    #print 'SRC:\n<<<<<<<>>>>>>>\n%s<<<<<>>>>>>' % src  # dbg
-    return src
 
 
 #-------------------------------------------------------------------------------
 # The top-level MultiEngine client adaptor
 #-------------------------------------------------------------------------------
 
 
 _prop_warn = """\
 
 We are currently refactoring the task dependency system.  This might
 involve the removal of this method and other methods related to engine
 properties.  Please see the docstrings for IPython.kernel.TaskRejectError 
 for more information."""
 
 
 class IFullBlockingMultiEngineClient(Interface):
     pass
 
 
 class FullBlockingMultiEngineClient(InteractiveMultiEngineClient):
     """
     A blocking client to the `IMultiEngine` controller interface.
     
     This class allows users to use a set of engines for a parallel
     computation through the `IMultiEngine` interface.  In this interface,
     each engine has a specific id (an int) that is used to refer to the
     engine, run code on it, etc.
     """
     
     implements(
         IFullBlockingMultiEngineClient,
         IMultiEngineMapperFactory,
         IMapper
     )
     
     def __init__(self, smultiengine):
         self.smultiengine = smultiengine
         self.block = True
         self.targets = 'all'
     
     def _findBlock(self, block=None):
         if block is None:
             return self.block
         else:
             if block in (True, False):
                 return block
             else:
                 raise ValueError("block must be True or False")
     
     def _findTargets(self, targets=None):
         if targets is None:
             return self.targets
         else:
             if not isinstance(targets, (str,list,tuple,int)):
                 raise ValueError("targets must be a str, list, tuple or int")
             return targets
     
     def _findTargetsAndBlock(self, targets=None, block=None):
         return self._findTargets(targets), self._findBlock(block) 
-    
+
+    def _bcft(self, *args, **kwargs):
+        try:
+            result = blockingCallFromThread(*args, **kwargs)
+        except DeadReferenceError:
+            raise error.ConnectionError(
+                """A connection error has occurred in trying to connect to the
+                controller. This is usually caused by the controller dying or 
+                being restarted. To resolve this issue try recreating the 
+                multiengine client."""
+            )
+        else:
+            return result
+
     def _blockFromThread(self, function, *args, **kwargs):
         block = kwargs.get('block', None)
         if block is None:
             raise error.MissingBlockArgument("'block' keyword argument is missing")
-        result = blockingCallFromThread(function, *args, **kwargs)
+        result = self._bcft(function, *args, **kwargs)
         if not block:
             result = PendingResult(self, result)
         return result
     
     def get_pending_deferred(self, deferredID, block):
-        return blockingCallFromThread(self.smultiengine.get_pending_deferred, deferredID, block)
+        return self._bcft(self.smultiengine.get_pending_deferred, deferredID, block)
     
     def barrier(self, pendingResults):
         """Synchronize a set of `PendingResults`.
         
         This method is a synchronization primitive that waits for a set of
         `PendingResult` objects to complete.  More specifically, barier does
         the following.
         
         * The `PendingResult`s are sorted by result_id.
         * The `get_result` method is called for each `PendingResult` sequentially
           with block=True.
         * If a `PendingResult` gets a result that is an exception, it is 
           trapped and can be re-raised later by calling `get_result` again.
         * The `PendingResult`s are flushed from the controller.
                 
         After barrier has been called on a `PendingResult`, its results can 
         be retrieved by calling `get_result` again or accesing the `r` attribute
         of the instance.
         """
         
         # Convert to list for sorting and check class type 
         prList = list(pendingResults)
         for pr in prList:
             if not isinstance(pr, PendingResult):
                 raise error.NotAPendingResult("Objects passed to barrier must be PendingResult instances")
                             
         # Sort the PendingResults so they are in order
         prList.sort()
         # Block on each PendingResult object
         for pr in prList:
             try:
                 result = pr.get_result(block=True)
             except Exception:
                 pass
     
     def flush(self):
         """
         Clear all pending deferreds/results from the controller.
         
         For each `PendingResult` that is created by this client, the controller
         holds on to the result for that `PendingResult`.  This can be a problem
         if there are a large number of `PendingResult` objects that are created.
         
         Once the result of the `PendingResult` has been retrieved, the result
         is removed from the controller, but if a user doesn't get a result (
         they just ignore the `PendingResult`) the result is kept forever on the
         controller.  This method allows the user to clear out all un-retrieved
         results on the controller. 
         """
-        r = blockingCallFromThread(self.smultiengine.clear_pending_deferreds)
+        r = self._bcft(self.smultiengine.clear_pending_deferreds)
         return r
     
     clear_pending_results = flush
     
     #---------------------------------------------------------------------------
     # IEngineMultiplexer related methods
     #---------------------------------------------------------------------------
     
     def execute(self, lines, targets=None, block=None):
         """
         Execute code on a set of engines.
         
         :Parameters:
             lines : str
                 The Python code to execute as a string
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
-        result = blockingCallFromThread(self.smultiengine.execute, lines,
+        result = self._bcft(self.smultiengine.execute, lines,
             targets=targets, block=block)
         if block:
             result = ResultList(result)
         else:
             result = PendingResult(self, result)
             result.add_callback(wrapResultList)
         return result
     
     def push(self, namespace, targets=None, block=None):
         """
         Push a dictionary of keys and values to engines namespace.
         
         Each engine has a persistent namespace.  This method is used to push
         Python objects into that namespace.
         
         The objects in the namespace must be pickleable.
         
         :Parameters:
             namespace : dict
                 A dict that contains Python objects to be injected into
                 the engine persistent namespace.
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push, namespace,
             targets=targets, block=block)
     
     def pull(self, keys, targets=None, block=None):
         """
         Pull Python objects by key out of engines namespaces.
         
         :Parameters:
             keys : str or list of str
                 The names of the variables to be pulled
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull, keys, targets=targets, block=block)
     
     def push_function(self, namespace, targets=None, block=None):
         """
         Push a Python function to an engine.
         
         This method is used to push a Python function to an engine.  This
         method can then be used in code on the engines.  Closures are not supported.
         
         :Parameters:
             namespace : dict
                 A dict whose values are the functions to be pushed.  The keys give
                 that names that the function will appear as in the engines
                 namespace.
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push_function, namespace, targets=targets, block=block)
     
     def pull_function(self, keys, targets=None, block=None):
         """
         Pull a Python function from an engine.
         
         This method is used to pull a Python function from an engine.
         Closures are not supported.
         
         :Parameters:
             keys : str or list of str
                 The names of the functions to be pulled
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull_function, keys, targets=targets, block=block)
     
     def push_serialized(self, namespace, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push_serialized, namespace, targets=targets, block=block)
     
     def pull_serialized(self, keys, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull_serialized, keys, targets=targets, block=block)
     
     def get_result(self, i=None, targets=None, block=None):
         """
         Get a previous result.
         
         When code is executed in an engine, a dict is created and returned.  This
         method retrieves that dict for previous commands.
         
         :Parameters:
             i : int
                 The number of the result to get
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
-        result = blockingCallFromThread(self.smultiengine.get_result, i, targets=targets, block=block)
+        result = self._bcft(self.smultiengine.get_result, i, targets=targets, block=block)
         if block:
             result = ResultList(result)
         else:
             result = PendingResult(self, result)
             result.add_callback(wrapResultList)
         return result
     
     def reset(self, targets=None, block=None):
         """
         Reset an engine.
         
         This method clears out the namespace of an engine.
 
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.reset, targets=targets, block=block)
     
     def keys(self, targets=None, block=None):
         """
         Get a list of all the variables in an engine's namespace.
 
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.keys, targets=targets, block=block)
     
     def kill(self, controller=False, targets=None, block=None):
         """
         Kill the engines and controller.
         
         This method is used to stop the engine and controller by calling
         `reactor.stop`.
         
         :Parameters:
             controller : boolean
                 If True, kill the engines and controller.  If False, just the 
                 engines
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.kill, controller, targets=targets, block=block)
     
     def clear_queue(self, targets=None, block=None):
         """
         Clear out the controller's queue for an engine.
         
         The controller maintains a queue for each engine.  This clear it out.
         
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.clear_queue, targets=targets, block=block)
     
     def queue_status(self, targets=None, block=None):
         """
         Get the status of an engines queue.
         
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.queue_status, targets=targets, block=block)
     
     def set_properties(self, properties, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.set_properties, properties, targets=targets, block=block)
     
     def get_properties(self, keys=None, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.get_properties, keys, targets=targets, block=block)
     
     def has_properties(self, keys, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.has_properties, keys, targets=targets, block=block)
     
     def del_properties(self, keys, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.del_properties, keys, targets=targets, block=block)
     
     def clear_properties(self, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.clear_properties, targets=targets, block=block)
     
     #---------------------------------------------------------------------------
     # IMultiEngine related methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         """
         Returns the ids of currently registered engines.
         """
-        result = blockingCallFromThread(self.smultiengine.get_ids)
+        result = self._bcft(self.smultiengine.get_ids)
         return result
         
     #---------------------------------------------------------------------------
     # IMultiEngineCoordinator
     #---------------------------------------------------------------------------
              
     def scatter(self, key, seq, dist='b', flatten=False, targets=None, block=None):
         """
         Partition a Python sequence and send the partitions to a set of engines.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.scatter, key, seq, 
             dist, flatten, targets=targets, block=block)
     
     def gather(self, key, dist='b', targets=None, block=None):
         """
         Gather a partitioned sequence on a set of engines as a single local seq.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.gather, key, dist, 
             targets=targets, block=block)
     
     def raw_map(self, func, seq, dist='b', targets=None, block=None):
         """
         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`.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.raw_map, func, seq, 
             dist, targets=targets, block=block)
     
     def map(self, func, *sequences):
         """
         A parallel version of Python's builtin `map` function.
         
         This method applies a function to sequences of arguments.  It 
         follows the same syntax as the builtin `map`.
         
         This method creates a mapper objects by calling `self.mapper` with
         no arguments and then uses that mapper to do the mapping.  See
         the documentation of `mapper` for more details.
         """
         return self.mapper().map(func, *sequences)
     
     def mapper(self, dist='b', targets='all', block=None):
         """
         Create a mapper object that has a `map` method.
         
         This method returns an object that implements the `IMapper` 
         interface.  This method is a factory that is used to control how 
         the map happens.
         
         :Parameters:
             dist : str
                 What decomposition to use, 'b' is the only one supported
                 currently
             targets : str, int, sequence of ints
                 Which engines to use for the map
             block : boolean
                 Should calls to `map` block or not
         """
         return MultiEngineMapper(self, dist, targets, block)
     
     def parallel(self, dist='b', targets=None, block=None):
         """
         A decorator that turns a function into a parallel function.
         
         This can be used as:
         
         @parallel()
         def f(x, y)
             ...
         
         f(range(10), range(10))
         
         This causes f(0,0), f(1,1), ... to be called in parallel.
         
         :Parameters:
             dist : str
                 What decomposition to use, 'b' is the only one supported
                 currently
             targets : str, int, sequence of ints
                 Which engines to use for the map
             block : boolean
                 Should calls to `map` block or not
         """
         targets, block = self._findTargetsAndBlock(targets, block)       
         mapper = self.mapper(dist, targets, block)
         pf = ParallelFunction(mapper)
         return pf
     
     #---------------------------------------------------------------------------
     # IMultiEngineExtras
     #---------------------------------------------------------------------------
     
     def zip_pull(self, keys, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.zip_pull, keys, 
             targets=targets, block=block)
     
     def run(self, filename, targets=None, block=None):
         """
         Run a Python code in a file on the engines.
         
         :Parameters:
             filename : str
                 The name of the local file to run
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.run, filename,
             targets=targets, block=block)
     
     def benchmark(self, push_size=10000):
         """
         Run performance benchmarks for the current IPython cluster.
         
         This method tests both the latency of sending command and data to the
         engines as well as the throughput of sending large objects to the
         engines using push.  The latency is measured by having one or more
         engines execute the command 'pass'.  The throughput is measure by 
         sending an NumPy array of size `push_size` to one or more engines.
         
         These benchmarks will vary widely on different hardware and networks
         and thus can be used to get an idea of the performance characteristics
         of a particular configuration of an IPython controller and engines.
         
         This function is not testable within our current testing framework.
         """
         import timeit, __builtin__
         __builtin__._mec_self = self
         benchmarks = {}
         repeat = 3
         count = 10
 
         timer = timeit.Timer('_mec_self.execute("pass",0)')
         result = 1000*min(timer.repeat(repeat,count))/count
         benchmarks['single_engine_latency'] = (result,'msec')
 
         timer = timeit.Timer('_mec_self.execute("pass")')
         result = 1000*min(timer.repeat(repeat,count))/count
         benchmarks['all_engine_latency'] = (result,'msec')
 
         try:
             import numpy as np
         except:
             pass
         else:
             timer = timeit.Timer(
                 "_mec_self.push(d)",
                 "import numpy as np; d = dict(a=np.zeros(%r,dtype='float64'))" % push_size
             )
             result = min(timer.repeat(repeat,count))/count
             benchmarks['all_engine_push'] = (1e-6*push_size*8/result, 'MB/sec')
 
         try:
             import numpy as np
         except:
             pass
         else:
             timer = timeit.Timer(
                 "_mec_self.push(d,0)",
                 "import numpy as np; d = dict(a=np.zeros(%r,dtype='float64'))" % push_size
             )
             result = min(timer.repeat(repeat,count))/count
             benchmarks['single_engine_push'] = (1e-6*push_size*8/result, 'MB/sec')
 
         return benchmarks
 
 
 components.registerAdapter(FullBlockingMultiEngineClient,
             IFullSynchronousMultiEngine, IFullBlockingMultiEngineClient)
 
 
 
 

IPython/kernel/taskclient.py

@@ -1,180 +1,194 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.tests.test_taskcontrollerxmlrpc -*-
 
 """
 A blocking version of the task client.
 """
 
 __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 zope.interface import Interface, implements
 from twisted.python import components
+from foolscap import DeadReferenceError
 
 from IPython.kernel.twistedutil import blockingCallFromThread
-from IPython.kernel import task
+from IPython.kernel import task, error
 from IPython.kernel.mapper import (
     SynchronousTaskMapper,
     ITaskMapperFactory,
     IMapper
 )
 from IPython.kernel.parallelfunction import (
     ParallelFunction, 
     ITaskParallelDecorator
 )
 
 #-------------------------------------------------------------------------------
 # The task client
 #-------------------------------------------------------------------------------
 
 class IBlockingTaskClient(Interface):
     """
     A vague interface of the blocking task client
     """
     pass
 
 class BlockingTaskClient(object):
     """
     A blocking task client that adapts a non-blocking one.
     """
     
     implements(
         IBlockingTaskClient,
         ITaskMapperFactory,
         IMapper,
         ITaskParallelDecorator
     )
     
     def __init__(self, task_controller):
         self.task_controller = task_controller
         self.block = True
-        
+
+    def _bcft(self, *args, **kwargs):
+        try:
+            result = blockingCallFromThread(*args, **kwargs)
+        except DeadReferenceError:
+            raise error.ConnectionError(
+                """A connection error has occurred in trying to connect to the
+                controller. This is usually caused by the controller dying or 
+                being restarted. To resolve this issue try recreating the 
+                task client."""
+            )
+        else:
+            return result
+
     def run(self, task, block=False):
         """Run a task on the `TaskController`.
         
         See the documentation of the `MapTask` and `StringTask` classes for
         details on how to build a task of different types.
         
         :Parameters:
             task : an `ITask` implementer
         
         :Returns: The int taskid of the submitted task.  Pass this to 
             `get_task_result` to get the `TaskResult` object.
         """
-        tid = blockingCallFromThread(self.task_controller.run, task)
+        tid = self._bcft(self.task_controller.run, task)
         if block:
             return self.get_task_result(tid, block=True)
         else:
             return tid
     
     def get_task_result(self, taskid, block=False):
         """
         Get a task result by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be retrieved.
             block : boolean
                 Should I block until the task is done?
         
         :Returns: A `TaskResult` object that encapsulates the task result.
         """
-        return blockingCallFromThread(self.task_controller.get_task_result,
+        return self._bcft(self.task_controller.get_task_result,
             taskid, block)
     
     def abort(self, taskid):
         """
         Abort a task by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be aborted.
         """
-        return blockingCallFromThread(self.task_controller.abort, taskid)
+        return self._bcft(self.task_controller.abort, taskid)
     
     def barrier(self, taskids):
         """Block until a set of tasks are completed.
         
         :Parameters:
             taskids : list, tuple
                 A sequence of taskids to block on.
         """
-        return blockingCallFromThread(self.task_controller.barrier, taskids)
+        return self._bcft(self.task_controller.barrier, taskids)
     
     def spin(self):
         """
         Touch the scheduler, to resume scheduling without submitting a task.
         
         This method only needs to be called in unusual situations where the
         scheduler is idle for some reason. 
         """
-        return blockingCallFromThread(self.task_controller.spin)
+        return self._bcft(self.task_controller.spin)
     
     def queue_status(self, verbose=False):
         """
         Get a dictionary with the current state of the task queue.
         
         :Parameters:
             verbose : boolean
                 If True, return a list of taskids.  If False, simply give
                 the number of tasks with each status.
         
         :Returns:
             A dict with the queue status.
         """
-        return blockingCallFromThread(self.task_controller.queue_status, verbose)
+        return self._bcft(self.task_controller.queue_status, verbose)
     
     def clear(self):
         """
         Clear all previously run tasks from the task controller.
         
         This is needed because the task controller keep all task results
         in memory.  This can be a problem is there are many completed
         tasks.  Users should call this periodically to clean out these
         cached task results.
         """
-        return blockingCallFromThread(self.task_controller.clear)
+        return self._bcft(self.task_controller.clear)
     
     def map(self, func, *sequences):
         """
         Apply func to *sequences elementwise.  Like Python's builtin map.
         
         This version is load balanced.
         """
         return self.mapper().map(func, *sequences)
 
     def mapper(self, clear_before=False, clear_after=False, retries=0, 
                 recovery_task=None, depend=None, block=True):
         """
         Create an `IMapper` implementer with a given set of arguments.
         
         The `IMapper` created using a task controller is load balanced.
         
         See the documentation for `IPython.kernel.task.BaseTask` for 
         documentation on the arguments to this method.
         """
         return SynchronousTaskMapper(self, clear_before=clear_before, 
             clear_after=clear_after, retries=retries, 
             recovery_task=recovery_task, depend=depend, block=block)
     
     def parallel(self, clear_before=False, clear_after=False, retries=0, 
         recovery_task=None, depend=None, block=True):
         mapper = self.mapper(clear_before, clear_after, retries,
             recovery_task, depend, block)
         pf = ParallelFunction(mapper)
         return pf
 
 components.registerAdapter(BlockingTaskClient,
             task.ITaskController, IBlockingTaskClient)
 
 

IPython/kernel/tests/test_enginefc.py

@@ -1,92 +1,92 b''
 # encoding: utf-8
 
 """This file contains unittests for the enginepb.py module."""
 
 __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
 #-------------------------------------------------------------------------------
 
 # Tell nose to skip this module
 __test__ = {}
 
 from twisted.python import components
 from twisted.internet import reactor, defer
 from twisted.spread import pb
 from twisted.internet.base import DelayedCall
 DelayedCall.debug = True
 
 import zope.interface as zi
 
 from IPython.kernel.fcutil import Tub, UnauthenticatedTub
 from IPython.kernel import engineservice as es
 from IPython.testing.util import DeferredTestCase
 from IPython.kernel.controllerservice import IControllerBase
 from IPython.kernel.enginefc import FCRemoteEngineRefFromService, IEngineBase
 from IPython.kernel.engineservice import IEngineQueued
 from IPython.kernel.engineconnector import EngineConnector
 
 from IPython.kernel.tests.engineservicetest import \
     IEngineCoreTestCase, \
     IEngineSerializedTestCase, \
     IEngineQueuedTestCase
 
 
 class EngineFCTest(DeferredTestCase, 
                  IEngineCoreTestCase, 
                  IEngineSerializedTestCase,
                  IEngineQueuedTestCase
                  ):
  
   zi.implements(IControllerBase)
  
   def setUp(self):
  
       # Start a server and append to self.servers
       self.controller_reference = FCRemoteEngineRefFromService(self)
       self.controller_tub = Tub()
-      self.controller_tub.listenOn('tcp:10105:interface=127.0.0.1')
-      self.controller_tub.setLocation('127.0.0.1:10105')
+      self.controller_tub.listenOn('tcp:10111:interface=127.0.0.1')
+      self.controller_tub.setLocation('127.0.0.1:10111')
      
       furl = self.controller_tub.registerReference(self.controller_reference)
       self.controller_tub.startService()
      
       # Start an EngineService and append to services/client
       self.engine_service = es.EngineService()
       self.engine_service.startService()
       self.engine_tub = Tub()
       self.engine_tub.startService()
       engine_connector = EngineConnector(self.engine_tub)
       d = engine_connector.connect_to_controller(self.engine_service, furl)
       # This deferred doesn't fire until after register_engine has returned and
       # thus, self.engine has been defined and the tets can proceed.
       return d
  
   def tearDown(self):
       dlist = []
       # Shut down the engine
       d = self.engine_tub.stopService()
       dlist.append(d)
       # Shut down the controller
       d = self.controller_tub.stopService()
       dlist.append(d)
       return defer.DeferredList(dlist)
  
   #---------------------------------------------------------------------------
   # Make me look like a basic controller
   #---------------------------------------------------------------------------
  
   def register_engine(self, engine_ref, id=None, ip=None, port=None, pid=None):
       self.engine = IEngineQueued(IEngineBase(engine_ref))
       return {'id':id}
  
   def unregister_engine(self, id):
       pass
  No newline at end of file

IPython/kernel/tests/test_multienginefc.py

@@ -1,152 +1,146 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 __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
 #-------------------------------------------------------------------------------
 
 # Tell nose to skip this module
 __test__ = {}
 
 from twisted.internet import defer, reactor
 
 from IPython.kernel.fcutil import Tub, UnauthenticatedTub
 
 from IPython.testing.util import DeferredTestCase
 from IPython.kernel.controllerservice import ControllerService
 from IPython.kernel.multiengine import IMultiEngine
 from IPython.kernel.tests.multienginetest import IFullSynchronousMultiEngineTestCase
 from IPython.kernel.multienginefc import IFCSynchronousMultiEngine
 from IPython.kernel import multiengine as me
-from IPython.kernel.clientconnector import ClientConnector
+from IPython.kernel.clientconnector import AsyncClientConnector
 from IPython.kernel.parallelfunction import ParallelFunction
 from IPython.kernel.error import CompositeError
 from IPython.kernel.util import printer
 
 
 def _raise_it(f):
     try:
         f.raiseException()
     except CompositeError, e:
         e.raise_exception()
 
 
-class FullSynchronousMultiEngineTestCase(DeferredTestCase, IFullSynchronousMultiEngineTestCase):
+class FullSynchronousMultiEngineTestCase(
+    DeferredTestCase, IFullSynchronousMultiEngineTestCase):
 
-    # XXX (fperez) this is awful: I'm fully disabling this entire test class.
-    # Right now it's blocking the tests from running at all, and I don't know
-    # how to fix it.  I hope Brian can have a stab at it, but at least by doing
-    # this we can run the entire suite to completion.
-    # Once the problem is cleared, remove this skip method.
-    skip = True
-    # END XXX
-    
     def setUp(self):
-    
+
         self.engines = []
-            
+
         self.controller = ControllerService()
         self.controller.startService()
         self.imultiengine = IMultiEngine(self.controller)
         self.mec_referenceable = IFCSynchronousMultiEngine(self.imultiengine)
 
         self.controller_tub = Tub()
-        self.controller_tub.listenOn('tcp:10105:interface=127.0.0.1')
-        self.controller_tub.setLocation('127.0.0.1:10105')
-    
+        self.controller_tub.listenOn('tcp:10111:interface=127.0.0.1')
+        self.controller_tub.setLocation('127.0.0.1:10111')
+
         furl = self.controller_tub.registerReference(self.mec_referenceable)
         self.controller_tub.startService()
-    
-        self.client_tub = ClientConnector()
-        d = self.client_tub.get_multiengine_client(furl)
+
+        self.client_tub = AsyncClientConnector()
+        d = self.client_tub.get_multiengine_client(furl_or_file=furl)
         d.addCallback(self.handle_got_client)
         return d
-    
+
     def handle_got_client(self, client):
         self.multiengine = client
 
     def tearDown(self):
         dlist = []
         # Shut down the multiengine client
         d = self.client_tub.tub.stopService()
         dlist.append(d)
         # Shut down the engines
         for e in self.engines:
             e.stopService()
         # Shut down the controller
         d = self.controller_tub.stopService()
         d.addBoth(lambda _: self.controller.stopService())
         dlist.append(d)
         return defer.DeferredList(dlist)
 
     def test_mapper(self):
         self.addEngine(4)
         m = self.multiengine.mapper()
         self.assertEquals(m.multiengine,self.multiengine)
         self.assertEquals(m.dist,'b')
         self.assertEquals(m.targets,'all')
         self.assertEquals(m.block,True)
-    
+
     def test_map_default(self):
         self.addEngine(4)
         m = self.multiengine.mapper()
         d = m.map(lambda x: 2*x, range(10))
         d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
         d.addCallback(lambda _: self.multiengine.map(lambda x: 2*x, range(10)))
         d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
         return d
-    
+
     def test_map_noblock(self):
         self.addEngine(4)
         m = self.multiengine.mapper(block=False)
         d = m.map(lambda x: 2*x, range(10))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
         return d
-            
+
     def test_mapper_fail(self):
         self.addEngine(4)
         m = self.multiengine.mapper()
         d = m.map(lambda x: 1/0, range(10))
         d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
         return d
-    
+
     def test_parallel(self):
         self.addEngine(4)
         p = self.multiengine.parallel()
         self.assert_(isinstance(p, ParallelFunction))
         @p
         def f(x): return 2*x
         d = f(range(10))
         d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
         return d
-    
+
     def test_parallel_noblock(self):
         self.addEngine(1)
         p = self.multiengine.parallel(block=False)
         self.assert_(isinstance(p, ParallelFunction))
         @p
         def f(x): return 2*x
         d = f(range(10))
         d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
         d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
         return d
-    
+
     def test_parallel_fail(self):
         self.addEngine(4)
         p = self.multiengine.parallel()
         self.assert_(isinstance(p, ParallelFunction))
         @p
         def f(x): return 1/0
         d = f(range(10))
         d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
         return d
+

IPython/kernel/tests/test_taskfc.py

@@ -1,169 +1,162 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 __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
 #-------------------------------------------------------------------------------
 
 # Tell nose to skip this module
 __test__ = {}
 
 import time
 
 from twisted.internet import defer, reactor
 
 from IPython.kernel.fcutil import Tub, UnauthenticatedTub
 
 from IPython.kernel import task as taskmodule
 from IPython.kernel import controllerservice as cs
 import IPython.kernel.multiengine as me
 from IPython.testing.util import DeferredTestCase
 from IPython.kernel.multienginefc import IFCSynchronousMultiEngine
 from IPython.kernel.taskfc import IFCTaskController
 from IPython.kernel.util import printer
 from IPython.kernel.tests.tasktest import ITaskControllerTestCase
-from IPython.kernel.clientconnector import ClientConnector
+from IPython.kernel.clientconnector import AsyncClientConnector
 from IPython.kernel.error import CompositeError
 from IPython.kernel.parallelfunction import ParallelFunction
 
 
 #-------------------------------------------------------------------------------
 # Tests
 #-------------------------------------------------------------------------------
 
 def _raise_it(f):
     try:
         f.raiseException()
     except CompositeError, e:
         e.raise_exception()
 
 class TaskTest(DeferredTestCase, ITaskControllerTestCase):
 
-    # XXX (fperez) this is awful: I'm fully disabling this entire test class.
-    # Right now it's blocking the tests from running at all, and I don't know
-    # how to fix it.  I hope Brian can have a stab at it, but at least by doing
-    # this we can run the entire suite to completion.
-    # Once the problem is cleared, remove this skip method.
-    skip = True
-    # END XXX
- 
     def setUp(self):
-    
+
         self.engines = []
-            
+
         self.controller = cs.ControllerService()
         self.controller.startService()
         self.imultiengine = me.IMultiEngine(self.controller)
         self.itc = taskmodule.ITaskController(self.controller)
         self.itc.failurePenalty = 0
-    
+
         self.mec_referenceable = IFCSynchronousMultiEngine(self.imultiengine)
         self.tc_referenceable = IFCTaskController(self.itc)
-    
+
         self.controller_tub = Tub()
-        self.controller_tub.listenOn('tcp:10105:interface=127.0.0.1')
-        self.controller_tub.setLocation('127.0.0.1:10105')
-    
+        self.controller_tub.listenOn('tcp:10111:interface=127.0.0.1')
+        self.controller_tub.setLocation('127.0.0.1:10111')
+
         mec_furl = self.controller_tub.registerReference(self.mec_referenceable)
         tc_furl = self.controller_tub.registerReference(self.tc_referenceable)
         self.controller_tub.startService()
-    
-        self.client_tub = ClientConnector()
-        d = self.client_tub.get_multiengine_client(mec_furl)
+
+        self.client_tub = AsyncClientConnector()
+        d = self.client_tub.get_multiengine_client(furl_or_file=mec_furl)
         d.addCallback(self.handle_mec_client)
-        d.addCallback(lambda _: self.client_tub.get_task_client(tc_furl))
+        d.addCallback(lambda _: self.client_tub.get_task_client(furl_or_file=tc_furl))
         d.addCallback(self.handle_tc_client)
         return d
-    
+
     def handle_mec_client(self, client):
         self.multiengine = client
 
     def handle_tc_client(self, client):
         self.tc = client
 
     def tearDown(self):
         dlist = []
         # Shut down the multiengine client
         d = self.client_tub.tub.stopService()
         dlist.append(d)
         # Shut down the engines
         for e in self.engines:
             e.stopService()
         # Shut down the controller
         d = self.controller_tub.stopService()
         d.addBoth(lambda _: self.controller.stopService())
         dlist.append(d)
         return defer.DeferredList(dlist)
-    
+
     def test_mapper(self):
         self.addEngine(1)
         m = self.tc.mapper()
         self.assertEquals(m.task_controller,self.tc)
         self.assertEquals(m.clear_before,False)
         self.assertEquals(m.clear_after,False)
         self.assertEquals(m.retries,0)
         self.assertEquals(m.recovery_task,None)
         self.assertEquals(m.depend,None)
         self.assertEquals(m.block,True)
-    
+
     def test_map_default(self):
         self.addEngine(1)
         m = self.tc.mapper()
         d = m.map(lambda x: 2*x, range(10))
         d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
         d.addCallback(lambda _: self.tc.map(lambda x: 2*x, range(10)))
         d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
         return d
-    
+
     def test_map_noblock(self):
         self.addEngine(1)
         m = self.tc.mapper(block=False)
         d = m.map(lambda x: 2*x, range(10))
         d.addCallback(lambda r: self.assertEquals(r,[x for x in range(10)]))
         return d
-            
+
     def test_mapper_fail(self):
         self.addEngine(1)
         m = self.tc.mapper()
         d = m.map(lambda x: 1/0, range(10))
         d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
         return d
-    
+
     def test_parallel(self):
         self.addEngine(1)
         p = self.tc.parallel()
         self.assert_(isinstance(p, ParallelFunction))
         @p
         def f(x): return 2*x
         d = f(range(10))
         d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
         return d
-    
+
     def test_parallel_noblock(self):
         self.addEngine(1)
         p = self.tc.parallel(block=False)
         self.assert_(isinstance(p, ParallelFunction))
         @p
         def f(x): return 2*x
         d = f(range(10))
         d.addCallback(lambda r: self.assertEquals(r,[x for x in range(10)]))
         return d
-    
+
     def test_parallel_fail(self):
         self.addEngine(1)
         p = self.tc.parallel()
         self.assert_(isinstance(p, ParallelFunction))
         @p
         def f(x): return 1/0
         d = f(range(10))
         d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
         return d
+