Work on engine/client reconnect logic.

IPython/kernel/clientconnector.py

@@ -1,655 +1,713 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
+from IPython.kernel.fcutil import (
+    Tub,
+    find_furl,
+    is_valid_furl_or_file,
+    validate_furl_or_file,
+    FURLError
+)
 from IPython.kernel.clusterdir import ClusterDir, ClusterDirError
 from IPython.kernel.launcher import IPClusterLauncher
-from IPython.kernel.twistedutil import gatherBoth, make_deferred
-from IPython.kernel.twistedutil import blockingCallFromThread
-
+from IPython.kernel.twistedutil import (
+    gatherBoth, 
+    make_deferred,
+    blockingCallFromThread, 
+    sleep_deferred
+)
 from IPython.utils.importstring import import_item
 from IPython.utils.genutils import get_ipython_dir
 
 from twisted.internet import defer
-from twisted.python import failure
+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,
                    ipythondir=None):
         """Find a FURL file by profile+ipythondir or cluster dir.
 
-        This raises an exception if a FURL file can't be found.
+        This raises an :exc:`~IPython.kernel.fcutil.FURLError` exception 
+        if a FURL file can't be found.
         """
         # Try by furl_or_file
         if furl_or_file is not None:
-            try:
-                furl = find_furl(furl_or_file)
-            except ValueError:
-                return furl
+            validate_furl_or_file(furl_or_file)
+            return furl_or_file
 
         if furl_file_name is None:
-            raise ValueError('A furl_file_name must be provided')
+            raise FURLError('A furl_file_name must be provided')
 
         # Try by cluster_dir
         if cluster_dir is not None:
             cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
             sdir = cluster_dir_obj.security_dir
             furl_file = os.path.join(sdir, furl_file_name)
-            return find_furl(furl_file)
+            validate_furl_or_file(furl_file)
+            return furl_file
 
         # Try by profile
         if ipythondir is None:
             ipythondir = get_ipython_dir()
         if profile is not None:
             cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
                 ipythondir, profile)
             sdir = cluster_dir_obj.security_dir
             furl_file = os.path.join(sdir, furl_file_name)
-            return find_furl(furl_file)
+            validate_furl_or_file(furl_file)
+            return furl_file
 
-        raise ValueError('Could not find a valid 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
+            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 = find_furl(furl_or_file)
+        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, ipythondir=None):
+                        furl_or_file=None, ipythondir=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 ipythondir and then in the directories
             listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
         furl_or_file : str
             A furl or a filename containing a FURLK. This is useful if you 
             simply know the location of the FURL file.
         ipythondir : str
             The location of the ipythondir if different from the default.
             This is used if the cluster directory is being found by profile.
 
         Returns
         -------
         A deferred to the actual client class.
         """
         return self.get_client(
             profile, cluster_dir, furl_or_file, 
-            'ipcontroller-tc.furl', ipythondir
+            'ipcontroller-tc.furl', ipythondir,
+            delay, max_tries
         )
 
     def get_multiengine_client(self, profile='default', cluster_dir=None,
-                               furl_or_file=None, ipythondir=None):
+                               furl_or_file=None, ipythondir=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 ipythondir and then in the directories
             listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
         furl_or_file : str
             A furl or a filename containing a FURLK. This is useful if you 
             simply know the location of the FURL file.
         ipythondir : str
             The location of the ipythondir if different from the default.
             This is used if the cluster directory is being found by profile.
                 
         Returns
         -------
         A deferred to the actual client class.
         """
         return self.get_client(
             profile, cluster_dir, furl_or_file, 
-            'ipcontroller-mec.furl', ipythondir
+            'ipcontroller-mec.furl', ipythondir,
+            delay, max_tries
         )
     
     def get_client(self, profile='default', cluster_dir=None,
-                   furl_or_file=None, furl_file_name=None, ipythondir=None):
+                   furl_or_file=None, furl_file_name=None, ipythondir=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 ipythondir and then in the directories
             listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
         furl_or_file : str
-            A furl or a filename containing a FURLK. This is useful if you 
+            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.
         ipythondir : str
             The location of the ipythondir if different from the default.
             This is used if the cluster directory is being found by profile.
 
         Returns
         -------
-        A deferred to the actual client class.
+        A deferred to the actual client class. Or a failure to a 
+        :exc:`FURLError`.
         """
         try:
-            furl = self._find_furl(
+            furl_file = self._find_furl(
                 profile, cluster_dir, furl_or_file,
                 furl_file_name, ipythondir
             )
-        except:
+        except FURLError:
             return defer.fail(failure.Failure())
 
-        d = self.get_reference(furl)
-
         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)
         return d
 
+    @inlineCallbacks
+    def _try_to_connect(self, furl_or_file, delay, max_tries, attempt):
+        """Try to connect to the controller with retry logic."""
+        if attempt < max_tries:
+            log.msg("Connecting to controller [%r]: %s" % \
+                (attempt, furl_or_file))
+            try:
+                self.furl = find_furl(furl_or_file)
+                # Uncomment this to see the FURL being tried.
+                # log.msg("FURL: %s" % self.furl)
+                rr = yield self.get_reference(self.furl)
+            except:
+                if attempt==max_tries-1:
+                    # This will propagate the exception all the way to the top
+                    # where it can be handled.
+                    raise
+                else:
+                    yield sleep_deferred(delay)
+                    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, ipythondir=None):
+                        furl_or_file=None, ipythondir=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 ipythondir and then in the directories
             listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
         furl_or_file : str
             A furl or a filename containing a FURLK. This is useful if you 
             simply know the location of the FURL file.
         ipythondir : str
             The location of the ipythondir if different from the default.
             This is used if the cluster directory is being found by profile.
 
         Returns
         -------
         The task client instance.
         """
         client = blockingCallFromThread(
             self.async_cc.get_task_client, profile, cluster_dir,
-            furl_or_file, ipythondir
+            furl_or_file, ipythondir, delay, max_tries
         )
         return client.adapt_to_blocking_client()
 
     def get_multiengine_client(self, profile='default', cluster_dir=None,
-                               furl_or_file=None, ipythondir=None):
+                               furl_or_file=None, ipythondir=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 ipythondir and then in the directories
             listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
         furl_or_file : str
             A furl or a filename containing a FURLK. This is useful if you 
             simply know the location of the FURL file.
         ipythondir : str
             The location of the ipythondir if different from the default.
             This is used if the cluster directory is being found by profile.
 
         Returns
         -------
         The multiengine client instance.
         """
         client = blockingCallFromThread(
             self.async_cc.get_multiengine_client, profile, cluster_dir,
-            furl_or_file, ipythondir
+            furl_or_file, ipythondir, delay, max_tries
         )
         return client.adapt_to_blocking_client()
 
     def get_client(self, profile='default', cluster_dir=None,
-                   furl_or_file=None, ipythondir=None):
+                   furl_or_file=None, ipythondir=None,
+                   delay=DELAY, max_tries=MAX_TRIES):
         client = blockingCallFromThread(
             self.async_cc.get_client, profile, cluster_dir,
-            furl_or_file, ipythondir
+            furl_or_file, ipythondir,
+            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, ipythondir=None,
                  auto_create=False, auto_stop=True):
         """Create a class to manage an IPython cluster.
 
         This class calls the :command:`ipcluster` command with the right
         options to start an IPython cluster.  Typically a cluster directory
         must be created (:command:`ipcluster create`) and configured before 
         using this class. Configuration is done by editing the 
         configuration files in the top level of the cluster directory.
 
         Parameters
         ----------
         profile : str
             The name of a cluster directory profile (default="default"). The
             cluster directory "cluster_<profile>" will be searched for
             in ``os.getcwd()``, the ipythondir and then in the directories
             listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
-        furl_or_file : str
-            A furl or a filename containing a FURLK. This is useful if you 
-            simply know the location of the FURL file.
         ipythondir : str
             The location of the ipythondir if different from the default.
             This is used if the cluster directory is being found by profile.
         auto_create : bool
             Automatically create the cluster directory it is dones't exist.
             This will usually only make sense if using a local cluster 
             (default=False).
         auto_stop : bool
             Automatically stop the cluster when this instance is garbage 
             collected (default=True).  This is useful if you want the cluster
             to live beyond your current process. There is also an instance
             attribute ``auto_stop`` to change this behavior.
         """
         self._setup_cluster_dir(profile, cluster_dir, ipythondir, auto_create)
         self.state = 'before'
         self.launcher = None
         self.client_connector = None
         self.auto_stop = auto_stop
 
     def __del__(self):
         if self.auto_stop and self.state=='running':
             print "Auto stopping the cluster..."
             self.stop()
 
     @property
     def location(self):
         if hasattr(self, 'cluster_dir_obj'):
             return self.cluster_dir_obj.location
         else:
             return ''
 
     @property
     def running(self):
         if self.state=='running':
             return True
         else:
             return False
 
     def _setup_cluster_dir(self, profile, cluster_dir, ipythondir, auto_create):
         if ipythondir is None:
             ipythondir = get_ipython_dir()
         if cluster_dir is not None:
             try:
                 self.cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
             except ClusterDirError:
                 pass
         if profile is not None:
             try:
                 self.cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
                     ipythondir, profile)
             except ClusterDirError:
                 pass
         if auto_create or profile=='default':
             # This should call 'ipcluster create --profile default
             self.cluster_dir_obj = ClusterDir.create_cluster_dir_by_profile(
                 ipythondir, profile)
         else:
             raise ClusterDirError('Cluster dir not found.')
 
     @make_deferred
     def start(self, n=2):
         """Start the IPython cluster with n engines.
 
         Parameters
         ----------
         n : int
             The number of engine to start.
         """
         # We might want to add logic to test if the cluster has started
         # by another process....
         if not self.state=='running':
             self.launcher = IPClusterLauncher(os.getcwd())
             self.launcher.ipcluster_n = n
             self.launcher.ipcluster_subcommand = 'start'
             d = self.launcher.start()
             d.addCallback(self._handle_start)
             return d
         else:
             raise ClusterStateError('Cluster is already running')
 
     @make_deferred
     def stop(self):
         """Stop the IPython cluster if it is running."""
         if self.state=='running':
             d1 = self.launcher.observe_stop()
             d1.addCallback(self._handle_stop)
             d2 = self.launcher.stop()
             return gatherBoth([d1, d2], consumeErrors=True)
         else:
             raise ClusterStateError("Cluster not running")
 
-    def get_multiengine_client(self):
+    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
+            cluster_dir=self.cluster_dir_obj.location,
+            delay=delay, max_tries=max_tries
         )
 
-    def get_task_client(self):
+    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
+            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, ipythondir=None,
                  auto_create=False, auto_stop=True):
         """Create a class to manage an IPython cluster.
 
         This class calls the :command:`ipcluster` command with the right
         options to start an IPython cluster.  Typically a cluster directory
         must be created (:command:`ipcluster create`) and configured before 
         using this class. Configuration is done by editing the 
         configuration files in the top level of the cluster directory.
 
         Parameters
         ----------
         profile : str
             The name of a cluster directory profile (default="default"). The
             cluster directory "cluster_<profile>" will be searched for
             in ``os.getcwd()``, the ipythondir and then in the directories
             listed in the :env:`IPCLUSTERDIR_PATH` environment variable.
         cluster_dir : str
             The full path to a cluster directory.  This is useful if profiles
             are not being used.
-        furl_or_file : str
-            A furl or a filename containing a FURLK. This is useful if you 
-            simply know the location of the FURL file.
         ipythondir : str
             The location of the ipythondir if different from the default.
             This is used if the cluster directory is being found by profile.
         auto_create : bool
             Automatically create the cluster directory it is dones't exist.
             This will usually only make sense if using a local cluster 
             (default=False).
         auto_stop : bool
             Automatically stop the cluster when this instance is garbage 
             collected (default=True).  This is useful if you want the cluster
             to live beyond your current process. There is also an instance
             attribute ``auto_stop`` to change this behavior.
         """
         self.async_cluster = AsyncCluster(
             profile, cluster_dir, ipythondir, auto_create, auto_stop
         )
         self.cluster_dir_obj = self.async_cluster.cluster_dir_obj
         self.client_connector = None
 
     def _set_auto_stop(self, value):
         self.async_cluster.auto_stop = value
 
     def _get_auto_stop(self):
         return self.async_cluster.auto_stop
 
     auto_stop = property(_get_auto_stop, _set_auto_stop)
 
     @property
     def location(self):
         return self.async_cluster.location
 
     @property
     def running(self):
         return self.async_cluster.running
 
     def start(self, n=2):
         """Start the IPython cluster with n engines.
 
         Parameters
         ----------
         n : int
             The number of engine to start.
         """
         return blockingCallFromThread(self.async_cluster.start, n)
 
     def stop(self):
         """Stop the IPython cluster if it is running."""
         return blockingCallFromThread(self.async_cluster.stop)
 
-    def get_multiengine_client(self):
+    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 = ClientConnector()
         return self.client_connector.get_multiengine_client(
-            cluster_dir=self.cluster_dir_obj.location
+            cluster_dir=self.cluster_dir_obj.location,
+            delay=delay, max_tries=max_tries
         )
 
-    def get_task_client(self):
+    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 = ClientConnector()
         return self.client_connector.get_task_client(
-            cluster_dir=self.cluster_dir_obj.location
+            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/engineconnector.py

@@ -1,130 +1,139 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 """A class that manages the engines connection 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
 #-----------------------------------------------------------------------------
 
 import os
 import cPickle as pickle
 
 from twisted.python import log, failure
 from twisted.internet import defer
 from twisted.internet.defer import inlineCallbacks, returnValue
 
-from IPython.kernel.fcutil import find_furl
+from IPython.kernel.fcutil import find_furl, validate_furl_or_file
 from IPython.kernel.enginefc import IFCEngine
-from IPython.kernel.twistedutil import sleep_deferred
+from IPython.kernel.twistedutil import sleep_deferred, make_deferred
 
 #-----------------------------------------------------------------------------
 # The ClientConnector class
 #-----------------------------------------------------------------------------
 
 
 class EngineConnectorError(Exception):
     pass
 
 
 class EngineConnector(object):
     """Manage an engines connection to a controller.
     
     This class takes a foolscap `Tub` and provides a `connect_to_controller` 
     method that will use the `Tub` to connect to a controller and register
     the engine with the controller.
     """
     
     def __init__(self, tub):
         self.tub = tub
 
+    @make_deferred
     def connect_to_controller(self, engine_service, furl_or_file,
                               delay=0.1, max_tries=10):
         """
         Make a connection to a controller specified by a furl.
         
         This method takes an `IEngineBase` instance and a foolcap URL and uses
         the `tub` attribute to make a connection to the controller.  The 
         foolscap URL contains all the information needed to connect to the 
         controller, including the ip and port as well as any encryption and 
         authentication information needed for the connection.
 
         After getting a reference to the controller, this method calls the 
         `register_engine` method of the controller to actually register the 
         engine.
 
         This method will try to connect to the controller multiple times with
         a delay in between.  Each time the FURL file is read anew.
 
         Parameters
         __________
         engine_service : IEngineBase
             An instance of an `IEngineBase` implementer
         furl_or_file : str
             A furl or a filename containing a furl
         delay : float
             The intial time to wait between connection attempts.  Subsequent
             attempts have increasing delays.
         max_tries : int
             The maximum number of connection attempts.
+
+        Returns
+        -------
+        A deferred to the registered client or a failure to an error
+        like :exc:`FURLError`.
         """
         if not self.tub.running:
             self.tub.startService()
         self.engine_service = engine_service
         self.engine_reference = IFCEngine(self.engine_service)
 
+        validate_furl_or_file(furl_or_file)
         d = self._try_to_connect(furl_or_file, delay, max_tries, attempt=0)
+        d.addCallback(self._register)
         return d
 
     @inlineCallbacks
     def _try_to_connect(self, furl_or_file, delay, max_tries, attempt):
         """Try to connect to the controller with retry logic."""
         if attempt < max_tries:
             log.msg("Attempting to connect to controller [%r]: %s" % \
                 (attempt, furl_or_file))
             try:
                 self.furl = find_furl(furl_or_file)
                 # Uncomment this to see the FURL being tried.
                 # log.msg("FURL: %s" % self.furl)
                 rr = yield self.tub.getReference(self.furl)
             except:
                 if attempt==max_tries-1:
                     # This will propagate the exception all the way to the top
                     # where it can be handled.
                     raise
                 else:
                     yield sleep_deferred(delay)
-                    yield self._try_to_connect(
+                    rr = yield self._try_to_connect(
                         furl_or_file, 1.5*delay, max_tries, attempt+1
                     )
+                    # rr becomes an int when there is a connection!!!
+                    returnValue(rr)
             else:
-                result = yield self._register(rr)
-                returnValue(result)
+                returnValue(rr)
         else:
             raise EngineConnectorError(
                 'Could not connect to controller, max_tries (%r) exceeded. '
                 'This usually means that i) the controller was not started, '
                 'or ii) a firewall was blocking the engine from connecting '
                 'to the controller.' % max_tries
             )
 
     def _register(self, rr):
         self.remote_ref = rr
         # Now register myself with the controller
         desired_id = self.engine_service.id
         d = self.remote_ref.callRemote('register_engine', self.engine_reference, 
             desired_id, os.getpid(), pickle.dumps(self.engine_service.properties,2))
         return d.addCallback(self._reference_sent)
 
     def _reference_sent(self, registration_dict):
         self.engine_service.id = registration_dict['id']
         log.msg("engine registration succeeded, got id: %r" % self.engine_service.id)
         return self.engine_service.id
 

IPython/kernel/fcutil.py

@@ -1,239 +1,266 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
 
 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
 
 #-----------------------------------------------------------------------------
 # 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."""
     if os.path.isfile(furl_file):
         f = open(furl_file, 'r')
         oldfurl = f.read().strip()
         f.close()
         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 furl.startswith("pb://"):
             return True
         elif furl.startswith("pbu://"):
             return False
     else:
-        raise ValueError("invalid FURL: %s" % furl)
+        raise FURLError("invalid FURL: %s" % furl)
 
 
 def is_valid(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
 
 
 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):
         with open(furl_or_file, 'r') as f:
             furl = f.read().strip()
         if is_valid(furl):
             return furl
-    raise ValueError("Not a FURL or a file containing a FURL: %s" % furl_or_file)
+    raise FURLError("Not a valid FURL or FURL file: %s" % 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
+
+
+def validate_furl_or_file(furl_or_file):
+    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:
                 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.
         """
         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/twistedutil.py

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