Initial refactor of task dependency system.

We are thinking about refactoring the task dependency system.  Currently
is is based on engine properties and a function sent with each task.

In this commit, I have added a TaskRejectError that tasks can raise to
indicate that the engine doesn't have the required dependencies.  For
now I have not removed any of the old stuff, but I have added warnings
that we might remove the older *_prop* methods in IMultiEngine.

See this ticket for more info: https://bugs.launchpad.net/bugs/361419

docs/examples/kernel/taskreject.py

@@ -0,0 +1,53 b''
+#!/usr/bin/env python
+# encoding: utf-8
+
+"""
+A new example showing how to use `TaskRejectError` to handle dependencies
+in the IPython task system.
+
+To run this example, do::
+
+    $ ipcluster local -n 4
+
+Then, in another terminal start up IPython and do::
+
+    In [0]: %run taskreject.py
+
+    In [1]: mec.execute('run=True', targets=[0,1])
+
+After the first command, the scheduler will keep rescheduling the tasks, as
+they will fail with `TaskRejectError`.  But after the second command, there
+are two engines that the tasks can run on.  The tasks are quickly funneled
+to these engines.
+
+If you want to see how the controller is scheduling and retrying the tasks
+do a `tail -f` on the controller's log file in ~/.ipython/log.
+"""
+
+#-----------------------------------------------------------------------------
+#  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.
+#-----------------------------------------------------------------------------
+
+from IPython.kernel import client
+from IPython.kernel import TaskRejectError
+
+mec = client.MultiEngineClient()
+tc = client.TaskClient()
+
+mec.execute('from IPython.kernel import TaskRejectError')
+mec.execute('run = False')
+
+def map_task():
+    if not run:
+        raise TaskRejectError('task dependency not met')
+    return 3.0e8
+
+task_ids = []
+
+for i in range(10):
+    task = client.MapTask(map_task, retries=20)
+    task_ids.append(tc.run(task, block=False))
+

IPython/kernel/__init__.py

@@ -1,23 +1,25 b''
 # encoding: utf-8
 """The IPython1 kernel.
 
 The IPython kernel actually refers to three things:
 
  * The IPython Engine
  * The IPython Controller
  * Clients to the IPython Controller
 
 The kernel module implements the engine, controller and client and all the 
 network protocols needed for the various entities to talk to each other.
 
 An end user should probably begin by looking at the `client.py` module
 if they need blocking clients or in `asyncclient.py` if they want asynchronous,
 deferred/Twisted using clients.
 """
 __docformat__ = "restructuredtext en"
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
+
+from IPython.kernel.error import TaskRejectError
  No newline at end of file

IPython/kernel/error.py

@@ -1,188 +1,202 b''
 # encoding: utf-8
 
 """Classes and functions for kernel related errors and exceptions."""
 
 __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 IPython.kernel.core import error
 from twisted.python import failure
 
 #-------------------------------------------------------------------------------
 # Error classes
 #-------------------------------------------------------------------------------
 
 class KernelError(error.IPythonError):
     pass
 
 class NotDefined(KernelError):
     def __init__(self, name):
         self.name = name
         self.args = (name,)
 
     def __repr__(self):
         return '<NotDefined: %s>' % self.name
     
     __str__ = __repr__
 
 class QueueCleared(KernelError):
     pass
 
 class IdInUse(KernelError):
     pass
 
 class ProtocolError(KernelError):
     pass
 
 class ConnectionError(KernelError):
     pass
 
 class InvalidEngineID(KernelError):
     pass
     
 class NoEnginesRegistered(KernelError):
     pass
     
 class InvalidClientID(KernelError):
     pass
     
 class InvalidDeferredID(KernelError):
     pass
     
 class SerializationError(KernelError):
     pass
     
 class MessageSizeError(KernelError):
     pass
     
 class PBMessageSizeError(MessageSizeError):
     pass
     
 class ResultNotCompleted(KernelError):
     pass
     
 class ResultAlreadyRetrieved(KernelError):
     pass
     
 class ClientError(KernelError):
     pass
 
 class TaskAborted(KernelError):
     pass
 
 class TaskTimeout(KernelError):
     pass
 
 class NotAPendingResult(KernelError):
     pass
 
 class UnpickleableException(KernelError):
     pass
 
 class AbortedPendingDeferredError(KernelError):
     pass
 
 class InvalidProperty(KernelError):
     pass
 
 class MissingBlockArgument(KernelError):
     pass
 
 class StopLocalExecution(KernelError):
     pass
 
 class SecurityError(KernelError):
     pass
 
 class FileTimeoutError(KernelError):
     pass
 
+class TaskRejectError(KernelError):
+    """Exception to raise when a task should be rejected by an engine.
+    
+    This exception can be used to allow a task running on an engine to test
+    if the engine (or the user's namespace on the engine) has the needed
+    task dependencies.  If not, the task should raise this exception.  For
+    the task to be retried on another engine, the task should be created
+    with the `retries` argument > 1.
+    
+    The advantage of this approach over our older properties system is that
+    tasks have full access to the user's namespace on the engines and the
+    properties don't have to be managed or tested by the controller.
+    """
+
 class CompositeError(KernelError):
     def __init__(self, message, elist):
         Exception.__init__(self, *(message, elist))
         self.message = message
         self.elist = elist
   
     def _get_engine_str(self, ev):
         try:
             ei = ev._ipython_engine_info
         except AttributeError:
             return '[Engine Exception]'
         else:
             return '[%i:%s]: ' % (ei['engineid'], ei['method'])
     
     def _get_traceback(self, ev):
         try:
             tb = ev._ipython_traceback_text
         except AttributeError:
             return 'No traceback available'
         else:
             return tb
   
     def __str__(self):
         s = str(self.message)
         for et, ev, etb in self.elist:
             engine_str = self._get_engine_str(ev)
             s = s + '\n' + engine_str + str(et.__name__) + ': ' + str(ev)
         return s
     
     def print_tracebacks(self, excid=None):
         if excid is None:
             for (et,ev,etb) in self.elist:
                 print self._get_engine_str(ev)
                 print self._get_traceback(ev)
                 print
         else:
             try:
                 et,ev,etb = self.elist[excid]
             except:
                 raise IndexError("an exception with index %i does not exist"%excid)
             else:
                 print self._get_engine_str(ev)
                 print self._get_traceback(ev)
     
     def raise_exception(self, excid=0):
         try:
             et,ev,etb = self.elist[excid]
         except:
             raise IndexError("an exception with index %i does not exist"%excid)
         else:
             raise et, ev, etb
 
 def collect_exceptions(rlist, method):
     elist = []
     for r in rlist:
         if isinstance(r, failure.Failure):
             r.cleanFailure()
             et, ev, etb = r.type, r.value, r.tb
             # Sometimes we could have CompositeError in our list.  Just take
             # the errors out of them and put them in our new list.  This 
             # has the effect of flattening lists of CompositeErrors into one
             # CompositeError
             if et==CompositeError:
                 for e in ev.elist:
                     elist.append(e)
             else:
                 elist.append((et, ev, etb))
     if len(elist)==0:
         return rlist
     else:
         msg = "one or more exceptions from call to method: %s" % (method)
         # This silliness is needed so the debugger has access to the exception
         # instance (e in this case)
         try:
             raise CompositeError(msg, elist)
         except CompositeError, e:
             raise e
             
 

IPython/kernel/multiengineclient.py

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

docs/source/parallel/visionhpc.txt

@@ -1,246 +1,250 b''
 ==================================
  IPython/Vision Beam Pattern Demo
 ==================================
 
+.. note::
+
+    This page has not been updated to reflect the recent work on ipcluster.
+    This work makes it much easier to use IPython on a cluster.
 
 Installing and testing IPython at OSC systems
 =============================================
 
 All components were installed from source and I have my environment set up to
 include ~/usr/local in my various necessary paths ($PATH, $PYTHONPATH, etc).
 Other than a slow filesystem for unpacking tarballs, the install went without a
 hitch.  For each needed component, I just downloaded the source tarball,
 unpacked it via::
 
   tar xzf (or xjf if it's bz2) filename.tar.{gz,bz2}
 
 and then installed them (including IPython itself) with::
 
   cd dirname/  # path to unpacked tarball
   python setup.py install --prefix=~/usr/local/
 
 The components I installed are listed below.  For each one I give the main
 project link as well as a direct one to the file I actually dowloaded and used.
 
 - nose, used for testing:
 http://somethingaboutorange.com/mrl/projects/nose/
 http://somethingaboutorange.com/mrl/projects/nose/nose-0.10.3.tar.gz
 
 - Zope interface, used to declare interfaces in twisted and ipython.  Note:
 you must get this from the page linked below and not fro the defaul
 one(http://www.zope.org/Products/ZopeInterface) because the latter has an
 older version, it hasn't been updated in a long time.  This pypi link has
 the current release (3.4.1 as of this writing):
 http://pypi.python.org/pypi/zope.interface
 http://pypi.python.org/packages/source/z/zope.interface/zope.interface-3.4.1.tar.gz
 
 - pyopenssl, security layer used by foolscap.  Note: version 0.7 *must* be
 used: 
 http://sourceforge.net/projects/pyopenssl/
 http://downloads.sourceforge.net/pyopenssl/pyOpenSSL-0.6.tar.gz?modtime=1212595285&big_mirror=0
 
 
 - Twisted, used for all networking:
 http://twistedmatrix.com/trac/wiki/Downloads
 http://tmrc.mit.edu/mirror/twisted/Twisted/8.1/Twisted-8.1.0.tar.bz2
 
 - Foolscap, used for managing connections securely:
 http://foolscap.lothar.com/trac
 http://foolscap.lothar.com/releases/foolscap-0.3.1.tar.gz
 
 
 - IPython itself:
 http://ipython.scipy.org/
 http://ipython.scipy.org/dist/ipython-0.9.1.tar.gz
 
 
 I then ran the ipython test suite via::
 
   iptest -vv
 
 and it passed with only::
 
     ======================================================================
     ERROR: testGetResult_2
     ----------------------------------------------------------------------
     DirtyReactorAggregateError: Reactor was unclean.
     Selectables:
     <Negotiation #0 on 10105>
 
     ----------------------------------------------------------------------
     Ran 419 tests in 33.971s
 
     FAILED (SKIP=4, errors=1)
 
 In three more runs of the test suite I was able to reproduce this error
 sometimes but not always; for now I think we can move on but we need to
 investigate further.  Especially if we start seeing problems in real use (the
 test suite stresses the networking layer in particular ways that aren't
 necessarily typical of normal use).
 
 Next, I started an 8-engine cluster via::
 
     perez@opt-login01[~]> ipcluster -n 8
     Starting controller: Controller PID: 30845
     ^X      Starting engines:    Engines PIDs:   [30846, 30847, 30848, 30849,
     30850, 30851, 30852, 30853]
     Log files: /home/perez/.ipython/log/ipcluster-30845-*
 
     Your cluster is up and running.
 
     [... etc]
 
 and in a separate ipython session checked that the cluster is running and I can
 access all the engines::
 
     In [1]: from IPython.kernel import client
 
     In [2]: mec = client.MultiEngineClient()
 
     In [3]: mec.get_ids()
     Out[3]: [0, 1, 2, 3, 4, 5, 6, 7]
 
 and run trivial code in them (after importing the ``random`` module in all
 engines)::
 
     In [11]: mec.execute("x=random.randint(0,10)")
     Out[11]:
     <Results List>
     [0] In [3]: x=random.randint(0,10)
     [1] In [3]: x=random.randint(0,10)
     [2] In [3]: x=random.randint(0,10)
     [3] In [3]: x=random.randint(0,10)
     [4] In [3]: x=random.randint(0,10)
     [5] In [3]: x=random.randint(0,10)
     [6] In [3]: x=random.randint(0,10)
     [7] In [3]: x=random.randint(0,10)
 
     In [12]: mec.pull('x')
     Out[12]: [10, 0, 8, 10, 2, 9, 10, 7]
 
 
 We'll continue conducting more complex tests later, including instaling Vision
 locally and running the beam demo.
 
 
 Michel's original instructions
 ==============================
 
 I got a Vision network that reproduces the beam pattern demo working:
 
 .. image:: vision_beam_pattern.png
    :width: 400
    :target: vision_beam_pattern.png
    :align: center
 
 
 I created a package called beamPattern that provides the function run() in its
 __init__.py file.
 
 A subpackage beamPattern/VisionInterface provides Vision nodes for:
 
 - computing Elevation and Azimuth from a 3D vector
 
 - Reading .mat files
 
 - taking the results gathered from the engines and creating the output that a
   single engine would have had produced
 
 The Mec node connect to a controller. In my network it was local but an furl
 can be specified to connect to a remote controller.
 
 The PRun Func node is from the IPython library of nodes. the import statement
 is used to get the run function from the beamPattern package and bu puting
 "run" in the function entry of this node we push this function to the engines.
 In addition to the node will create input ports for all arguments of the
 function being pushed (i.e. the run function)
 
 The second input port on PRun Fun take an integer specifying the rank of the
 argument we want to scatter. All other arguments will be pushed to the engines.
 
 The ElevAzim node has a 3D vector widget and computes the El And Az values
 which are passed into the PRun Fun node through the ports created
 automatically. The Mat node allows to select the .mat file, reads it and passed
 the data to the locdata port created automatically on PRun Func
 
 The calculation is executed in parallel, and the results are gathered and
 output.  Instead of having a list of 3 vectors we nd up with a list of n*3
 vectors where n is the number of engines. unpackDectorResults will turn it into
 a list of 3. We then plot x, y, and 10*log10(z)
 
 
 Installation
 ------------
 
 - inflate beamPattern into the site-packages directory for the MGL tools.
 
 - place the appended IPythonNodes.py and StandardNodes.py into the Vision
   package of the MGL tools.
 
 - place the appended items.py in the NetworkEditor package of the MGL tools
 
 - run vision for the network beamPat5_net.py::
 
     vision beamPat5_net.py
 
 Once the network is running, you can:
 
 - double click on the MEC node and either use an emptty string for the furl to
   connect to a local engine or cut and paste the furl to the engine you want to
   use
 
 - click on the yellow lighting bold to run the network.
 
 - Try modifying the MAT file or change the Vector used top compute elevation
   and Azimut.
   
 
 Fernando's notes
 ================
 
 - I had to install IPython and all its dependencies for the python used by the
   MGL tools.
 
 - Then I had to install scipy 0.6.0 for it, since the nodes needed Scipy.  To
   do this I sourced the mglenv.sh script and then ran::
 
     python setup.py install --prefix=~/usr/opt/mgl
 
 
 Using PBS
 =========
 
 The following PBS script can be used to start the engines::
 
     #PBS -N bgranger-ipython
     #PBS -j oe
     #PBS -l walltime=00:10:00
     #PBS -l nodes=4:ppn=4
 
     cd $PBS_O_WORKDIR
     export PATH=$HOME/usr/local/bin
     export PYTHONPATH=$HOME/usr/local/lib/python2.4/site-packages
     /usr/local/bin/mpiexec -n 16 ipengine
 
 
 If this file is called ``ipython_pbs.sh``, then the in one login windows
 (i.e. on the head-node -- ``opt-login01.osc.edu``), run ``ipcontroller``. In
 another login window on the same node, run the above script::
 
     qsub ipython_pbs.sh
 
 If you look at the first window, you will see some diagnostic output
 from ipcontroller. You can then get the furl from your own
 ``~/.ipython/security`` directory and then connect to it remotely.
 
 You might need to set up an SSH tunnel, however; if this doesn't work as
 advertised::
 
     ssh -L 10115:localhost:10105 bic
 
 
 Links to other resources
 ========================
 
 - http://www.osc.edu/~unpingco/glenn_NewLynx2_Demo.avi