Fixing small things in response to review.

IPython/kernel/multiengine.py

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

IPython/kernel/scripts/ipcluster.py

@@ -1,519 +1,521 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 """Start an IPython cluster = (controller + engines)."""
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import re
 import sys
 import signal
 pjoin = os.path.join
 
 from twisted.internet import reactor, defer
 from twisted.internet.protocol import ProcessProtocol
-from twisted.python import failure, log
 from twisted.internet.error import ProcessDone, ProcessTerminated
 from twisted.internet.utils import getProcessOutput
+from twisted.python import failure, log
 
 from IPython.external import argparse
 from IPython.external import Itpl
-from IPython.kernel.twistedutil import gatherBoth
-from IPython.kernel.util import printer
 from IPython.genutils import get_ipython_dir, num_cpus
 from IPython.kernel.fcutil import have_crypto
 from IPython.kernel.error import SecurityError
+from IPython.kernel.fcutil import have_crypto
+from IPython.kernel.twistedutil import gatherBoth
+from IPython.kernel.util import printer
+
 
 #-----------------------------------------------------------------------------
 # General process handling code
 #-----------------------------------------------------------------------------
 
 def find_exe(cmd):
     try:
         import win32api
     except ImportError:
         raise ImportError('you need to have pywin32 installed for this to work')
     else:
-		try:
-			(path, offest) = win32api.SearchPath(os.environ['PATH'],cmd + '.exe')
-		except:
-			(path, offset) = win32api.SearchPath(os.environ['PATH'],cmd + '.bat')
+        try:
+            (path, offest) = win32api.SearchPath(os.environ['PATH'],cmd + '.exe')
+        except:
+            (path, offset) = win32api.SearchPath(os.environ['PATH'],cmd + '.bat')
     return path
 
 class ProcessStateError(Exception):
     pass
 
 class UnknownStatus(Exception):
     pass
 
 class LauncherProcessProtocol(ProcessProtocol):
     """
     A ProcessProtocol to go with the ProcessLauncher.
     """
     def __init__(self, process_launcher):
         self.process_launcher = process_launcher
     
     def connectionMade(self):
         self.process_launcher.fire_start_deferred(self.transport.pid)
     
     def processEnded(self, status):
         value = status.value
         if isinstance(value, ProcessDone):
             self.process_launcher.fire_stop_deferred(0)
         elif isinstance(value, ProcessTerminated):
             self.process_launcher.fire_stop_deferred(
                 {'exit_code':value.exitCode,
                  'signal':value.signal,
                  'status':value.status
                 }
             )
         else:
             raise UnknownStatus("unknown exit status, this is probably a bug in Twisted")
 
     def outReceived(self, data):
         log.msg(data)
 
     def errReceived(self, data):
         log.err(data)
 
 class ProcessLauncher(object):
     """
     Start and stop an external process in an asynchronous manner.
     
     Currently this uses deferreds to notify other parties of process state
     changes.  This is an awkward design and should be moved to using
     a formal NotificationCenter.
     """
     def __init__(self, cmd_and_args):
         self.cmd = cmd_and_args[0]
         self.args = cmd_and_args
         self._reset()
     
     def _reset(self):
         self.process_protocol = None
         self.pid = None
         self.start_deferred = None
         self.stop_deferreds = []
         self.state = 'before' # before, running, or after
     
     @property
     def running(self):
         if self.state == 'running':
             return True
         else:
             return False
     
     def fire_start_deferred(self, pid):
         self.pid = pid
         self.state = 'running'
         log.msg('Process %r has started with pid=%i' % (self.args, pid))
         self.start_deferred.callback(pid)
     
     def start(self):
         if self.state == 'before':
             self.process_protocol = LauncherProcessProtocol(self)
             self.start_deferred = defer.Deferred()
             self.process_transport = reactor.spawnProcess(
                 self.process_protocol,
                 self.cmd,
                 self.args,
                 env=os.environ
             )
             return self.start_deferred
         else:
             s = 'the process has already been started and has state: %r' % \
                 self.state
             return defer.fail(ProcessStateError(s))
     
     def get_stop_deferred(self):
         if self.state == 'running' or self.state == 'before':
             d = defer.Deferred()
             self.stop_deferreds.append(d)
             return d
         else:
             s = 'this process is already complete'
             return defer.fail(ProcessStateError(s))
     
     def fire_stop_deferred(self, exit_code):
         log.msg('Process %r has stopped with %r' % (self.args, exit_code))
         self.state = 'after'
         for d in self.stop_deferreds:
             d.callback(exit_code)
     
     def signal(self, sig):
         """
         Send a signal to the process.
         
         The argument sig can be ('KILL','INT', etc.) or any signal number.
         """
         if self.state == 'running':
             self.process_transport.signalProcess(sig)
     
     # def __del__(self):
     #     self.signal('KILL')
     
     def interrupt_then_kill(self, delay=1.0):
         self.signal('INT')
         reactor.callLater(delay, self.signal, 'KILL')
 
 
 #-----------------------------------------------------------------------------
 # Code for launching controller and engines
 #-----------------------------------------------------------------------------
 
 
 class ControllerLauncher(ProcessLauncher):
     
     def __init__(self, extra_args=None):
         if sys.platform == 'win32':
             # This logic is needed because the ipcontroller script doesn't
             # always get installed in the same way or in the same location.
             from IPython.kernel.scripts import ipcontroller
             script_location = ipcontroller.__file__.replace('.pyc', '.py')
             # The -u option here turns on unbuffered output, which is required
             # on Win32 to prevent wierd conflict and problems with Twisted
             args = [find_exe('python'), '-u', script_location]
         else:
             args = ['ipcontroller']
         self.extra_args = extra_args
         if extra_args is not None:
             args.extend(extra_args)
         
         ProcessLauncher.__init__(self, args)
 
 
 class EngineLauncher(ProcessLauncher):
     
     def __init__(self, extra_args=None):
         if sys.platform == 'win32':
             # This logic is needed because the ipcontroller script doesn't
             # always get installed in the same way or in the same location.
             from IPython.kernel.scripts import ipengine
             script_location = ipengine.__file__.replace('.pyc', '.py')
             # The -u option here turns on unbuffered output, which is required
             # on Win32 to prevent wierd conflict and problems with Twisted
             args = [find_exe('python'), '-u', script_location]
         else:
             args = ['ipengine']
         self.extra_args = extra_args
         if extra_args is not None:
             args.extend(extra_args)
         
         ProcessLauncher.__init__(self, args)
 
 
 class LocalEngineSet(object):
     
     def __init__(self, extra_args=None):
         self.extra_args = extra_args
         self.launchers = []
     
     def start(self, n):
         dlist = []
         for i in range(n):
             el = EngineLauncher(extra_args=self.extra_args)
             d = el.start()
             self.launchers.append(el)
             dlist.append(d)
         dfinal = gatherBoth(dlist, consumeErrors=True)
         dfinal.addCallback(self._handle_start)
         return dfinal
     
     def _handle_start(self, r):
         log.msg('Engines started with pids: %r' % r)
         return r
     
     def _handle_stop(self, r):
         log.msg('Engines received signal: %r' % r)
         return r
     
     def signal(self, sig):
         dlist = []
         for el in self.launchers:
             d = el.get_stop_deferred()
             dlist.append(d)
             el.signal(sig)
         dfinal = gatherBoth(dlist, consumeErrors=True)
         dfinal.addCallback(self._handle_stop)
         return dfinal
     
     def interrupt_then_kill(self, delay=1.0):
         dlist = []
         for el in self.launchers:
             d = el.get_stop_deferred()
             dlist.append(d)
             el.interrupt_then_kill(delay)
         dfinal = gatherBoth(dlist, consumeErrors=True)
         dfinal.addCallback(self._handle_stop)
         return dfinal
 
 
 class BatchEngineSet(object):
     
     # Subclasses must fill these in.  See PBSEngineSet
     submit_command = ''
     delete_command = ''
     job_id_regexp = ''
     
     def __init__(self, template_file, **kwargs):
         self.template_file = template_file
         self.context = {}
         self.context.update(kwargs)
         self.batch_file = self.template_file+'-run'
 
     def parse_job_id(self, output):
         m = re.match(self.job_id_regexp, output)
         if m is not None:
             job_id = m.group()
         else:
             raise Exception("job id couldn't be determined: %s" % output)
         self.job_id = job_id
         log.msg('Job started with job id: %r' % job_id)
         return job_id
     
     def write_batch_script(self, n):
         self.context['n'] = n
         template = open(self.template_file, 'r').read()
         log.msg('Using template for batch script: %s' % self.template_file)
         script_as_string = Itpl.itplns(template, self.context)
         log.msg('Writing instantiated batch script: %s' % self.batch_file)
         f = open(self.batch_file,'w')
         f.write(script_as_string)
         f.close()
-
+    
     def handle_error(self, f):
         f.printTraceback()
         f.raiseException()
     
     def start(self, n):
         self.write_batch_script(n)
         d = getProcessOutput(self.submit_command,
             [self.batch_file],env=os.environ)
         d.addCallback(self.parse_job_id)
         d.addErrback(self.handle_error)
         return d
-        
+    
     def kill(self):
         d = getProcessOutput(self.delete_command,
             [self.job_id],env=os.environ)
         return d
 
 class PBSEngineSet(BatchEngineSet):
     
     submit_command = 'qsub'
     delete_command = 'qdel'
     job_id_regexp = '\d+'
     
     def __init__(self, template_file, **kwargs):
         BatchEngineSet.__init__(self, template_file, **kwargs)
 
 
 #-----------------------------------------------------------------------------
 # Main functions for the different types of clusters
 #-----------------------------------------------------------------------------
 
 # TODO:
 # The logic in these codes should be moved into classes like LocalCluster
 # MpirunCluster, PBSCluster, etc.  This would remove alot of the duplications.
 # The main functions should then just parse the command line arguments, create
 # the appropriate class and call a 'start' method.
 
 def check_security(args, cont_args):
     if (not args.x or not args.y) and not have_crypto:
         log.err("""
 OpenSSL/pyOpenSSL is not available, so we can't run in secure mode.
 Try running ipcluster with the -xy flags:  ipcluster local -xy -n 4""")
         reactor.stop()
         return False
     if args.x:
         cont_args.append('-x')
     if args.y:
         cont_args.append('-y')
     return True
 
 def main_local(args):
     cont_args = []
     cont_args.append('--logfile=%s' % pjoin(args.logdir,'ipcontroller'))
 
     # Check security settings before proceeding
-    keep_going = check_security(args, cont_args)
-    if not keep_going: return
+    if not check_security(args, cont_args):
+        return
 
     cl = ControllerLauncher(extra_args=cont_args)
     dstart = cl.start()
     def start_engines(cont_pid):
         engine_args = []
         engine_args.append('--logfile=%s' % \
             pjoin(args.logdir,'ipengine%s-' % cont_pid))
         eset = LocalEngineSet(extra_args=engine_args)
         def shutdown(signum, frame):
             log.msg('Stopping local cluster')
             # We are still playing with the times here, but these seem
             # to be reliable in allowing everything to exit cleanly.
             eset.interrupt_then_kill(0.5)
             cl.interrupt_then_kill(0.5)
             reactor.callLater(1.0, reactor.stop)
         signal.signal(signal.SIGINT,shutdown)
         d = eset.start(args.n)
         return d
     def delay_start(cont_pid):
         # This is needed because the controller doesn't start listening
         # right when it starts and the controller needs to write
         # furl files for the engine to pick up
         reactor.callLater(1.0, start_engines, cont_pid)
     dstart.addCallback(delay_start)
     dstart.addErrback(lambda f: f.raiseException())
 
 def main_mpirun(args):
     cont_args = []
     cont_args.append('--logfile=%s' % pjoin(args.logdir,'ipcontroller'))
 
     # Check security settings before proceeding
-    keep_going = check_security(args, cont_args)
-    if not keep_going: return
+    if not check_security(args, cont_args):
+        return
 
     cl = ControllerLauncher(extra_args=cont_args)
     dstart = cl.start()
     def start_engines(cont_pid):
         raw_args = ['mpirun']
         raw_args.extend(['-n',str(args.n)])
         raw_args.append('ipengine')
         raw_args.append('-l')
         raw_args.append(pjoin(args.logdir,'ipengine%s-' % cont_pid))
         if args.mpi:
             raw_args.append('--mpi=%s' % args.mpi)
         eset = ProcessLauncher(raw_args)
         def shutdown(signum, frame):
             log.msg('Stopping local cluster')
             # We are still playing with the times here, but these seem
             # to be reliable in allowing everything to exit cleanly.
             eset.interrupt_then_kill(1.0)
             cl.interrupt_then_kill(1.0)
             reactor.callLater(2.0, reactor.stop)
         signal.signal(signal.SIGINT,shutdown)
         d = eset.start()
         return d
     def delay_start(cont_pid):
         # This is needed because the controller doesn't start listening
         # right when it starts and the controller needs to write
         # furl files for the engine to pick up
         reactor.callLater(1.0, start_engines, cont_pid)
     dstart.addCallback(delay_start)
     dstart.addErrback(lambda f: f.raiseException())
 
 def main_pbs(args):
     cont_args = []
     cont_args.append('--logfile=%s' % pjoin(args.logdir,'ipcontroller'))
 
     # Check security settings before proceeding
-    keep_going = check_security(args, cont_args)
-    if not keep_going: return
+    if not check_security(args, cont_args):
+        return
 
     cl = ControllerLauncher(extra_args=cont_args)
     dstart = cl.start()
     def start_engines(r):
         pbs_set =  PBSEngineSet(args.pbsscript)
         def shutdown(signum, frame):
             log.msg('Stopping pbs cluster')
             d = pbs_set.kill()
             d.addBoth(lambda _: cl.interrupt_then_kill(1.0))
             d.addBoth(lambda _: reactor.callLater(2.0, reactor.stop))
         signal.signal(signal.SIGINT,shutdown)
         d = pbs_set.start(args.n)
         return d
     dstart.addCallback(start_engines)
     dstart.addErrback(lambda f: f.raiseException())
 
 
 def get_args():
     base_parser = argparse.ArgumentParser(add_help=False)
     base_parser.add_argument(
         '-x',
         action='store_true',
         dest='x',
         help='turn off client security'
     )
     base_parser.add_argument(
         '-y',
         action='store_true',
         dest='y',
         help='turn off engine security'
     )
     base_parser.add_argument(
         "--logdir", 
         type=str,
         dest="logdir",
         help="directory to put log files (default=$IPYTHONDIR/log)",
         default=pjoin(get_ipython_dir(),'log')
     )
     base_parser.add_argument(
         "-n",
         "--num", 
         type=int,
         dest="n",
         default=2,
         help="the number of engines to start"
     )
     
     parser = argparse.ArgumentParser(
         description='IPython cluster startup.  This starts a controller and\
         engines using various approaches.  THIS IS A TECHNOLOGY PREVIEW AND\
         THE API WILL CHANGE SIGNIFICANTLY BEFORE THE FINAL RELEASE.'
     )
     subparsers = parser.add_subparsers(
         help='available cluster types.  For help, do "ipcluster TYPE --help"')
     
     parser_local = subparsers.add_parser(
         'local',
         help='run a local cluster',
         parents=[base_parser]
     )
     parser_local.set_defaults(func=main_local)
     
     parser_mpirun = subparsers.add_parser(
         'mpirun',
         help='run a cluster using mpirun',
         parents=[base_parser]
     )
     parser_mpirun.add_argument(
         "--mpi",
         type=str,
         dest="mpi", # Don't put a default here to allow no MPI support
         help="how to call MPI_Init (default=mpi4py)"
     )
     parser_mpirun.set_defaults(func=main_mpirun)
     
     parser_pbs = subparsers.add_parser(
         'pbs', 
         help='run a pbs cluster',
         parents=[base_parser]
     )
     parser_pbs.add_argument(
         '--pbs-script',
         type=str, 
         dest='pbsscript',
         help='PBS script template',
         default='pbs.template'
     )
     parser_pbs.set_defaults(func=main_pbs)
     args = parser.parse_args()
     return args
 
 def main():
     args = get_args()
     reactor.callWhenRunning(args.func, args)
     log.startLogging(sys.stdout)
     reactor.run()
 
 if __name__ == '__main__':
     main()