merge from trunk

IPython/kernel/mapper.py

@@ -0,0 +1,233 b''
+# encoding: utf-8
+
+"""A parallelized version of Python's builtin map."""
+
+__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 types import FunctionType
+from zope.interface import Interface, implements
+from IPython.kernel.task import MapTask
+from IPython.kernel.twistedutil import DeferredList, gatherBoth
+from IPython.kernel.util import printer
+from IPython.kernel.error import collect_exceptions
+
+#----------------------------------------------------------------------------
+# Code
+#----------------------------------------------------------------------------
+
+class IMapper(Interface):
+    """The basic interface for a Mapper.
+    
+    This defines a generic interface for mapping.  The idea of this is 
+    similar to that of Python's builtin `map` function, which applies a function
+    elementwise to a sequence.
+    """
+    
+    def map(func, *seqs):
+        """Do map in parallel.
+        
+        Equivalent to map(func, *seqs) or:
+        
+        [func(seqs[0][0], seqs[1][0],...), func(seqs[0][1], seqs[1][1],...),...]
+        
+        :Parameters:
+            func : FunctionType
+                The function to apply to the sequence
+            sequences : tuple of iterables
+                A sequence of iterables that are used for sucessive function
+                arguments.  This work just like map
+        """
+
+class IMultiEngineMapperFactory(Interface):
+    """
+    An interface for something that creates `IMapper` instances.
+    """
+    
+    def mapper(dist='b', targets='all', block=True):
+        """
+        Create an `IMapper` implementer with a given set of arguments.
+        
+        The `IMapper` created using a multiengine controller is 
+        not load balanced.
+        """
+
+class ITaskMapperFactory(Interface):
+    """
+    An interface for something that creates `IMapper` instances.
+    """
+    
+    def mapper(clear_before=False, clear_after=False, retries=0, 
+                recovery_task=None, depend=None, block=True):
+        """
+        Create an `IMapper` implementer with a given set of arguments.
+        
+        The `IMapper` created using a task controller is load balanced.
+        
+        See the documentation for `IPython.kernel.task.BaseTask` for 
+        documentation on the arguments to this method.
+        """
+
+
+class MultiEngineMapper(object):
+    """
+    A Mapper for `IMultiEngine` implementers.
+    """
+    
+    implements(IMapper)
+    
+    def __init__(self, multiengine, dist='b', targets='all', block=True):
+        """
+        Create a Mapper for a multiengine.
+        
+        The value of all arguments are used for all calls to `map`.  This
+        class allows these arguemnts to be set for a series of map calls.
+        
+        :Parameters:
+            multiengine : `IMultiEngine` implementer
+                The multiengine to use for running the map commands
+            dist : str
+                The type of decomposition to use.  Only block ('b') is
+                supported currently
+            targets : (str, int, tuple of ints)
+                The engines to use in the map
+            block : boolean
+                Whether to block when the map is applied
+        """
+        self.multiengine = multiengine
+        self.dist = dist
+        self.targets = targets
+        self.block = block
+    
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is not load balanced.
+        """
+        max_len = max(len(s) for s in sequences)
+        for s in sequences:
+            if len(s)!=max_len:
+                raise ValueError('all sequences must have equal length')
+        assert isinstance(func, (str, FunctionType)), "func must be a fuction or str"
+        return self.multiengine.raw_map(func, sequences, dist=self.dist,
+            targets=self.targets, block=self.block)
+
+class TaskMapper(object):
+    """
+    Make an `ITaskController` look like an `IMapper`.
+    
+    This class provides a load balanced version of `map`.
+    """
+    
+    def __init__(self, task_controller, clear_before=False, clear_after=False, retries=0, 
+            recovery_task=None, depend=None, block=True):
+        """
+        Create a `IMapper` given a `TaskController` and arguments.
+        
+        The additional arguments are those that are common to all types of 
+        tasks and are described in the documentation for 
+        `IPython.kernel.task.BaseTask`.
+        
+        :Parameters:
+            task_controller : an `IBlockingTaskClient` implementer
+                The `TaskController` to use for calls to `map`
+        """
+        self.task_controller = task_controller
+        self.clear_before = clear_before
+        self.clear_after = clear_after
+        self.retries = retries
+        self.recovery_task = recovery_task
+        self.depend = depend
+        self.block = block
+    
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is load balanced.
+        """
+        max_len = max(len(s) for s in sequences)
+        for s in sequences:
+            if len(s)!=max_len:
+                raise ValueError('all sequences must have equal length')
+        task_args = zip(*sequences)
+        task_ids = []
+        dlist = []
+        for ta in task_args:
+            task = MapTask(func, ta, clear_before=self.clear_before,
+                clear_after=self.clear_after, retries=self.retries,
+                recovery_task=self.recovery_task, depend=self.depend)
+            dlist.append(self.task_controller.run(task))
+        dlist = gatherBoth(dlist, consumeErrors=1)
+        dlist.addCallback(collect_exceptions,'map')
+        if self.block:
+            def get_results(task_ids):
+                d = self.task_controller.barrier(task_ids)
+                d.addCallback(lambda _: gatherBoth([self.task_controller.get_task_result(tid) for tid in task_ids], consumeErrors=1))
+                d.addCallback(collect_exceptions, 'map')
+                return d
+            dlist.addCallback(get_results)
+        return dlist
+
+class SynchronousTaskMapper(object):
+    """
+    Make an `IBlockingTaskClient` look like an `IMapper`.
+    
+    This class provides a load balanced version of `map`.
+    """
+    
+    def __init__(self, task_controller, clear_before=False, clear_after=False, retries=0, 
+            recovery_task=None, depend=None, block=True):
+        """
+        Create a `IMapper` given a `IBlockingTaskClient` and arguments.
+        
+        The additional arguments are those that are common to all types of 
+        tasks and are described in the documentation for 
+        `IPython.kernel.task.BaseTask`.
+        
+        :Parameters:
+            task_controller : an `IBlockingTaskClient` implementer
+                The `TaskController` to use for calls to `map`
+        """
+        self.task_controller = task_controller
+        self.clear_before = clear_before
+        self.clear_after = clear_after
+        self.retries = retries
+        self.recovery_task = recovery_task
+        self.depend = depend
+        self.block = block
+    
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is load balanced.
+        """
+        max_len = max(len(s) for s in sequences)
+        for s in sequences:
+            if len(s)!=max_len:
+                raise ValueError('all sequences must have equal length')
+        task_args = zip(*sequences)
+        task_ids = []
+        for ta in task_args:
+            task = MapTask(func, ta, clear_before=self.clear_before,
+                clear_after=self.clear_after, retries=self.retries,
+                recovery_task=self.recovery_task, depend=self.depend)
+            task_ids.append(self.task_controller.run(task))
+        if self.block:
+            self.task_controller.barrier(task_ids)
+            task_results = [self.task_controller.get_task_result(tid) for tid in task_ids]
+            return task_results
+        else:
+            return task_ids
  No newline at end of file

IPython/testing/plugin/Makefile

@@ -0,0 +1,20 b''
+# Set this prefix to where you want to install the plugin
+PREFIX=~/usr/local
+PREFIX=~/tmp/local
+
+plugin: IPython_doctest_plugin.egg-info
+
+test: plugin dtexample.py
+	nosetests -s --with-ipdoctest --doctest-tests --doctest-extension=txt \
+	dtexample.py test*.txt
+
+deb: plugin dtexample.py
+	nosetests -vs --with-ipdoctest --doctest-tests --doctest-extension=txt \
+        test_combo.txt
+
+IPython_doctest_plugin.egg-info: ipdoctest.py setup.py
+	python setup.py install --prefix=$(PREFIX)
+	touch $@
+
+clean:
+	rm -rf IPython_doctest_plugin.egg-info *~ *pyc build/ dist/

IPython/testing/plugin/README.txt

@@ -0,0 +1,39 b''
+=======================================================
+ Nose plugin with IPython and extension module support
+=======================================================
+
+This directory provides the key functionality for test support that IPython
+needs as a nose plugin, which can be installed for use in projects other than
+IPython.
+
+The presence of a Makefile here is mostly for development and debugging
+purposes as it only provides a few shorthand commands.  You can manually
+install the plugin by using standard Python procedures (``setup.py install``
+with appropriate arguments).
+
+To install the plugin using the Makefile, edit its first line to reflect where
+you'd like the installation.  If you want it system-wide, you may want to edit
+the install line in the plugin target to use sudo and no prefix::
+
+  sudo python setup.py install
+
+instead of the code using `--prefix` that's in there.
+
+Once you've set the prefix, simply build/install the plugin with::
+
+  make
+
+and run the tests with::
+
+  make test
+
+You should see output similar to::
+
+  maqroll[plugin]> make test
+  nosetests -s --with-ipdoctest --doctest-tests dtexample.py
+  ..
+  ----------------------------------------------------------------------
+  Ran 2 tests in 0.016s
+
+  OK
+

IPython/testing/plugin/dtexample.py

@@ -0,0 +1,72 b''
+"""Simple example using doctests.
+
+This file just contains doctests both using plain python and IPython prompts.
+All tests should be loaded by nose.
+"""
+
+def pyfunc():
+    """Some pure python tests...
+
+    >>> pyfunc()
+    'pyfunc'
+
+    >>> import os
+
+    >>> 2+3
+    5
+
+    >>> for i in range(3):
+    ...     print i,
+    ...     print i+1,
+    ...
+    0 1 1 2 2 3
+    """
+
+    return 'pyfunc'
+
+def ipfunc():
+    """Some ipython tests...
+
+    In [1]: import os
+
+    In [2]: cd /
+    /
+
+    In [3]: 2+3
+    Out[3]: 5
+
+    In [26]: for i in range(3):
+       ....:     print i,
+       ....:     print i+1,
+       ....:
+    0 1 1 2 2 3
+
+
+    Examples that access the operating system work:
+
+    In [1]: !echo hello
+    hello
+
+    In [2]: !echo hello > /tmp/foo
+
+    In [3]: !cat /tmp/foo
+    hello
+
+    In [4]: rm -f /tmp/foo
+
+    It's OK to use '_' for the last result, but do NOT try to use IPython's
+    numbered history of _NN outputs, since those won't exist under the
+    doctest environment:
+
+    In [7]: 3+4
+    Out[7]: 7
+
+    In [8]: _+3
+    Out[8]: 10
+
+    In [9]: ipfunc()
+    Out[9]: 'ipfunc'
+    """
+
+    return 'ipfunc'
+

IPython/testing/plugin/ipdoctest.py

@@ -0,0 +1,587 b''
+"""Nose Plugin that supports IPython doctests.
+
+Limitations:
+
+- When generating examples for use as doctests, make sure that you have
+  pretty-printing OFF.  This can be done either by starting ipython with the
+  flag '--nopprint', by setting pprint to 0 in your ipythonrc file, or by
+  interactively disabling it with %Pprint.  This is required so that IPython
+  output matches that of normal Python, which is used by doctest for internal
+  execution.
+
+- Do not rely on specific prompt numbers for results (such as using
+  '_34==True', for example).  For IPython tests run via an external process the
+  prompt numbers may be different, and IPython tests run as normal python code
+  won't even have these special _NN variables set at all.
+
+- IPython functions that produce output as a side-effect of calling a system
+  process (e.g. 'ls') can be doc-tested, but they must be handled in an
+  external IPython process.  Such doctests must be tagged with:
+
+        # ipdoctest: EXTERNAL
+
+  so that the testing machinery handles them differently.  Since these are run
+  via pexpect in an external process, they can't deal with exceptions or other
+  fancy featurs of regular doctests.  You must limit such tests to simple
+  matching of the output.  For this reason, I recommend you limit these kinds
+  of doctests to features that truly require a separate process, and use the
+  normal IPython ones (which have all the features of normal doctests) for
+  everything else.  See the examples at the bottom of this file for a
+  comparison of what can be done with both types.
+"""
+
+
+#-----------------------------------------------------------------------------
+# Module imports
+
+# From the standard library
+import __builtin__
+import commands
+import doctest
+import inspect
+import logging
+import os
+import re
+import sys
+import unittest
+
+from inspect import getmodule
+
+# Third-party modules
+import nose.core
+
+from nose.plugins import doctests, Plugin
+from nose.util import anyp, getpackage, test_address, resolve_name, tolist
+
+# Our own imports
+#from extdoctest import ExtensionDoctest, DocTestFinder
+#from dttools import DocTestFinder, DocTestCase
+#-----------------------------------------------------------------------------
+# Module globals and other constants
+
+log = logging.getLogger(__name__)
+
+###########################################################################
+# *** HACK ***
+# We must start our own ipython object and heavily muck with it so that all the
+# modifications IPython makes to system behavior don't send the doctest
+# machinery into a fit.  This code should be considered a gross hack, but it
+# gets the job done.
+
+def start_ipython():
+    """Start a global IPython shell, which we need for IPython-specific syntax.
+    """
+    import IPython
+
+    def xsys(cmd):
+        """Execute a command and print its output.
+
+        This is just a convenience function to replace the IPython system call
+        with one that is more doctest-friendly.
+        """
+        cmd = _ip.IP.var_expand(cmd,depth=1)
+        sys.stdout.write(commands.getoutput(cmd))
+        sys.stdout.flush()
+
+    # Store certain global objects that IPython modifies
+    _displayhook = sys.displayhook
+    _excepthook = sys.excepthook
+    _main = sys.modules.get('__main__')
+
+    # Start IPython instance
+    IPython.Shell.IPShell(['--classic','--noterm_title'])
+
+    # Deactivate the various python system hooks added by ipython for
+    # interactive convenience so we don't confuse the doctest system
+    sys.modules['__main__'] = _main
+    sys.displayhook = _displayhook
+    sys.excepthook = _excepthook
+
+    # So that ipython magics and aliases can be doctested (they work by making
+    # a call into a global _ip object)
+    _ip = IPython.ipapi.get()
+    __builtin__._ip = _ip
+
+    # Modify the IPython system call with one that uses getoutput, so that we
+    # can capture subcommands and print them to Python's stdout, otherwise the
+    # doctest machinery would miss them.
+    _ip.system = xsys
+
+# The start call MUST be made here.  I'm not sure yet why it doesn't work if
+# it is made later, at plugin initialization time, but in all my tests, that's
+# the case.
+start_ipython()
+
+# *** END HACK ***
+###########################################################################
+
+#-----------------------------------------------------------------------------
+# Modified version of the one in the stdlib, that fixes a python bug (doctests
+# not found in extension modules, http://bugs.python.org/issue3158)
+class DocTestFinder(doctest.DocTestFinder):
+
+    def _from_module(self, module, object):
+        """
+        Return true if the given object is defined in the given
+        module.
+        """
+        if module is None:
+            #print '_fm C1'  # dbg
+            return True
+        elif inspect.isfunction(object):
+            #print '_fm C2'  # dbg
+            return module.__dict__ is object.func_globals
+        elif inspect.isbuiltin(object):
+            #print '_fm C2-1'  # dbg
+            return module.__name__ == object.__module__
+        elif inspect.isclass(object):
+            #print '_fm C3'  # dbg
+            return module.__name__ == object.__module__
+        elif inspect.ismethod(object):
+            # This one may be a bug in cython that fails to correctly set the
+            # __module__ attribute of methods, but since the same error is easy
+            # to make by extension code writers, having this safety in place
+            # isn't such a bad idea
+            #print '_fm C3-1'  # dbg
+            return module.__name__ == object.im_class.__module__
+        elif inspect.getmodule(object) is not None:
+            #print '_fm C4'  # dbg
+            #print 'C4 mod',module,'obj',object # dbg
+            return module is inspect.getmodule(object)
+        elif hasattr(object, '__module__'):
+            #print '_fm C5'  # dbg
+            return module.__name__ == object.__module__
+        elif isinstance(object, property):
+            #print '_fm C6'  # dbg
+            return True # [XX] no way not be sure.
+        else:
+            raise ValueError("object must be a class or function")
+
+
+
+    def _find(self, tests, obj, name, module, source_lines, globs, seen):
+        """
+        Find tests for the given object and any contained objects, and
+        add them to `tests`.
+        """
+
+        doctest.DocTestFinder._find(self,tests, obj, name, module,
+                                    source_lines, globs, seen)
+
+        # Below we re-run pieces of the above method with manual modifications,
+        # because the original code is buggy and fails to correctly identify
+        # doctests in extension modules.
+
+        # Local shorthands
+        from inspect import isroutine, isclass, ismodule
+
+        # Look for tests in a module's contained objects.
+        if inspect.ismodule(obj) and self._recurse:
+            for valname, val in obj.__dict__.items():
+                valname1 = '%s.%s' % (name, valname)
+                if ( (isroutine(val) or isclass(val))
+                     and self._from_module(module, val) ):
+
+                    self._find(tests, val, valname1, module, source_lines,
+                               globs, seen)
+
+
+        # Look for tests in a class's contained objects.
+        if inspect.isclass(obj) and self._recurse:
+            #print 'RECURSE into class:',obj  # dbg
+            for valname, val in obj.__dict__.items():
+                #valname1 = '%s.%s' % (name, valname)  # dbg
+                #print 'N',name,'VN:',valname,'val:',str(val)[:77] # dbg
+                # Special handling for staticmethod/classmethod.
+                if isinstance(val, staticmethod):
+                    val = getattr(obj, valname)
+                if isinstance(val, classmethod):
+                    val = getattr(obj, valname).im_func
+
+                # Recurse to methods, properties, and nested classes.
+                if ((inspect.isfunction(val) or inspect.isclass(val) or
+                     inspect.ismethod(val) or
+                      isinstance(val, property)) and
+                      self._from_module(module, val)):
+                    valname = '%s.%s' % (name, valname)
+                    self._find(tests, val, valname, module, source_lines,
+                               globs, seen)
+
+
+class DocTestCase(doctests.DocTestCase):
+    """Proxy for DocTestCase: provides an address() method that
+    returns the correct address for the doctest case. Otherwise
+    acts as a proxy to the test case. To provide hints for address(),
+    an obj may also be passed -- this will be used as the test object
+    for purposes of determining the test address, if it is provided.
+    """
+
+    # doctests loaded via find(obj) omit the module name
+    # so we need to override id, __repr__ and shortDescription
+    # bonus: this will squash a 2.3 vs 2.4 incompatiblity
+    def id(self):
+        name = self._dt_test.name
+        filename = self._dt_test.filename
+        if filename is not None:
+            pk = getpackage(filename)
+            if pk is not None and not name.startswith(pk):
+                name = "%s.%s" % (pk, name)
+        return name
+
+
+# Classes and functions
+
+def is_extension_module(filename):
+    """Return whether the given filename is an extension module.
+
+    This simply checks that the extension is either .so or .pyd.
+    """
+    return os.path.splitext(filename)[1].lower() in ('.so','.pyd')
+
+
+# A simple subclassing of the original with a different class name, so we can
+# distinguish and treat differently IPython examples from pure python ones.
+class IPExample(doctest.Example): pass
+
+class IPExternalExample(doctest.Example):
+    """Doctest examples to be run in an external process."""
+
+    def __init__(self, source, want, exc_msg=None, lineno=0, indent=0,
+                 options=None):
+        # Parent constructor
+        doctest.Example.__init__(self,source,want,exc_msg,lineno,indent,options)
+
+        # An EXTRA newline is needed to prevent pexpect hangs
+        self.source += '\n'
+
+class IPDocTestParser(doctest.DocTestParser):
+    """
+    A class used to parse strings containing doctest examples.
+
+    Note: This is a version modified to properly recognize IPython input and
+    convert any IPython examples into valid Python ones.
+    """
+    # This regular expression is used to find doctest examples in a
+    # string.  It defines three groups: `source` is the source code
+    # (including leading indentation and prompts); `indent` is the
+    # indentation of the first (PS1) line of the source code; and
+    # `want` is the expected output (including leading indentation).
+
+    # Classic Python prompts or default IPython ones
+    _PS1_PY = r'>>>'
+    _PS2_PY = r'\.\.\.'
+
+    _PS1_IP = r'In\ \[\d+\]:'
+    _PS2_IP = r'\ \ \ \.\.\.+:'
+
+    _RE_TPL = r'''
+        # Source consists of a PS1 line followed by zero or more PS2 lines.
+        (?P<source>
+            (?:^(?P<indent> [ ]*) (?P<ps1> %s) .*)    # PS1 line
+            (?:\n           [ ]*  (?P<ps2> %s) .*)*)  # PS2 lines
+        \n? # a newline
+        # Want consists of any non-blank lines that do not start with PS1.
+        (?P<want> (?:(?![ ]*$)    # Not a blank line
+                     (?![ ]*%s)   # Not a line starting with PS1
+                     (?![ ]*%s)   # Not a line starting with PS2
+                     .*$\n?       # But any other line
+                  )*)
+                  '''
+
+    _EXAMPLE_RE_PY = re.compile( _RE_TPL % (_PS1_PY,_PS2_PY,_PS1_PY,_PS2_PY),
+                                 re.MULTILINE | re.VERBOSE)
+
+    _EXAMPLE_RE_IP = re.compile( _RE_TPL % (_PS1_IP,_PS2_IP,_PS1_IP,_PS2_IP),
+                                 re.MULTILINE | re.VERBOSE)
+
+    def ip2py(self,source):
+        """Convert input IPython source into valid Python."""
+        out = []
+        newline = out.append
+        for line in source.splitlines():
+            #newline(_ip.IPipython.prefilter(line,True))
+            newline(_ip.IP.prefilter(line,True))
+        newline('')  # ensure a closing newline, needed by doctest
+        return '\n'.join(out)
+
+    def parse(self, string, name='<string>'):
+        """
+        Divide the given string into examples and intervening text,
+        and return them as a list of alternating Examples and strings.
+        Line numbers for the Examples are 0-based.  The optional
+        argument `name` is a name identifying this string, and is only
+        used for error messages.
+        """
+
+        #print 'Parse string:\n',string # dbg
+
+        string = string.expandtabs()
+        # If all lines begin with the same indentation, then strip it.
+        min_indent = self._min_indent(string)
+        if min_indent > 0:
+            string = '\n'.join([l[min_indent:] for l in string.split('\n')])
+
+        output = []
+        charno, lineno = 0, 0
+
+        # Whether to convert the input from ipython to python syntax
+        ip2py = False
+        # Find all doctest examples in the string.  First, try them as Python
+        # examples, then as IPython ones
+        terms = list(self._EXAMPLE_RE_PY.finditer(string))
+        if terms:
+            # Normal Python example
+            #print '-'*70  # dbg
+            #print 'PyExample, Source:\n',string  # dbg
+            #print '-'*70  # dbg
+            Example = doctest.Example
+        else:
+            # It's an ipython example.  Note that IPExamples are run
+            # in-process, so their syntax must be turned into valid python.
+            # IPExternalExamples are run out-of-process (via pexpect) so they
+            # don't need any filtering (a real ipython will be executing them).
+            terms = list(self._EXAMPLE_RE_IP.finditer(string))
+            if re.search(r'#\s*ipdoctest:\s*EXTERNAL',string):
+                #print '-'*70  # dbg
+                #print 'IPExternalExample, Source:\n',string  # dbg
+                #print '-'*70  # dbg
+                Example = IPExternalExample
+            else:
+                #print '-'*70  # dbg
+                #print 'IPExample, Source:\n',string  # dbg
+                #print '-'*70  # dbg
+                Example = IPExample
+                ip2py = True
+
+        for m in terms:
+            # Add the pre-example text to `output`.
+            output.append(string[charno:m.start()])
+            # Update lineno (lines before this example)
+            lineno += string.count('\n', charno, m.start())
+            # Extract info from the regexp match.
+            (source, options, want, exc_msg) = \
+                     self._parse_example(m, name, lineno,ip2py)
+            if Example is IPExternalExample:
+                options[doctest.NORMALIZE_WHITESPACE] = True
+                want += '\n'
+            # Create an Example, and add it to the list.
+            if not self._IS_BLANK_OR_COMMENT(source):
+                #print 'Example source:', source # dbg
+                output.append(Example(source, want, exc_msg,
+                                      lineno=lineno,
+                                      indent=min_indent+len(m.group('indent')),
+                                      options=options))
+            # Update lineno (lines inside this example)
+            lineno += string.count('\n', m.start(), m.end())
+            # Update charno.
+            charno = m.end()
+        # Add any remaining post-example text to `output`.
+        output.append(string[charno:])
+
+        return output
+
+    def _parse_example(self, m, name, lineno,ip2py=False):
+        """
+        Given a regular expression match from `_EXAMPLE_RE` (`m`),
+        return a pair `(source, want)`, where `source` is the matched
+        example's source code (with prompts and indentation stripped);
+        and `want` is the example's expected output (with indentation
+        stripped).
+
+        `name` is the string's name, and `lineno` is the line number
+        where the example starts; both are used for error messages.
+
+        Optional:
+        `ip2py`: if true, filter the input via IPython to convert the syntax
+        into valid python.
+        """
+
+        # Get the example's indentation level.
+        indent = len(m.group('indent'))
+
+        # Divide source into lines; check that they're properly
+        # indented; and then strip their indentation & prompts.
+        source_lines = m.group('source').split('\n')
+
+        # We're using variable-length input prompts
+        ps1 = m.group('ps1')
+        ps2 = m.group('ps2')
+        ps1_len = len(ps1)
+
+        self._check_prompt_blank(source_lines, indent, name, lineno,ps1_len)
+        if ps2:
+            self._check_prefix(source_lines[1:], ' '*indent + ps2, name, lineno)
+
+        source = '\n'.join([sl[indent+ps1_len+1:] for sl in source_lines])
+
+        if ip2py:
+            # Convert source input from IPython into valid Python syntax
+            source = self.ip2py(source)
+
+        # Divide want into lines; check that it's properly indented; and
+        # then strip the indentation.  Spaces before the last newline should
+        # be preserved, so plain rstrip() isn't good enough.
+        want = m.group('want')
+        want_lines = want.split('\n')
+        if len(want_lines) > 1 and re.match(r' *$', want_lines[-1]):
+            del want_lines[-1]  # forget final newline & spaces after it
+        self._check_prefix(want_lines, ' '*indent, name,
+                           lineno + len(source_lines))
+
+        # Remove ipython output prompt that might be present in the first line
+        want_lines[0] = re.sub(r'Out\[\d+\]: \s*?\n?','',want_lines[0])
+
+        want = '\n'.join([wl[indent:] for wl in want_lines])
+
+        # If `want` contains a traceback message, then extract it.
+        m = self._EXCEPTION_RE.match(want)
+        if m:
+            exc_msg = m.group('msg')
+        else:
+            exc_msg = None
+
+        # Extract options from the source.
+        options = self._find_options(source, name, lineno)
+
+        return source, options, want, exc_msg
+
+    def _check_prompt_blank(self, lines, indent, name, lineno, ps1_len):
+        """
+        Given the lines of a source string (including prompts and
+        leading indentation), check to make sure that every prompt is
+        followed by a space character.  If any line is not followed by
+        a space character, then raise ValueError.
+
+        Note: IPython-modified version which takes the input prompt length as a
+        parameter, so that prompts of variable length can be dealt with.
+        """
+        space_idx = indent+ps1_len
+        min_len = space_idx+1
+        for i, line in enumerate(lines):
+            if len(line) >=  min_len and line[space_idx] != ' ':
+                raise ValueError('line %r of the docstring for %s '
+                                 'lacks blank after %s: %r' %
+                                 (lineno+i+1, name,
+                                  line[indent:space_idx], line))
+
+SKIP = doctest.register_optionflag('SKIP')
+
+###########################################################################
+
+class DocFileCase(doctest.DocFileCase):
+    """Overrides to provide filename
+    """
+    def address(self):
+        return (self._dt_test.filename, None, None)
+
+
+class ExtensionDoctest(doctests.Doctest):
+    """Nose Plugin that supports doctests in extension modules.
+    """
+    name = 'extdoctest'   # call nosetests with --with-extdoctest
+    enabled = True
+
+    def options(self, parser, env=os.environ):
+        Plugin.options(self, parser, env)
+
+    def configure(self, options, config):
+        Plugin.configure(self, options, config)
+        self.doctest_tests = options.doctest_tests
+        self.extension = tolist(options.doctestExtension)
+        self.finder = DocTestFinder()
+        self.parser = doctest.DocTestParser()
+
+
+    def loadTestsFromExtensionModule(self,filename):
+        bpath,mod = os.path.split(filename)
+        modname = os.path.splitext(mod)[0]
+        try:
+            sys.path.append(bpath)
+            module = __import__(modname)
+            tests = list(self.loadTestsFromModule(module))
+        finally:
+            sys.path.pop()
+        return tests
+
+    def loadTestsFromFile(self, filename):
+        if is_extension_module(filename):
+            for t in self.loadTestsFromExtensionModule(filename):
+                yield t
+        else:
+            ## for t in list(doctests.Doctest.loadTestsFromFile(self,filename)):
+            ##     yield t
+            pass
+
+        if self.extension and anyp(filename.endswith, self.extension):
+            #print 'lTF',filename  # dbg
+            name = os.path.basename(filename)
+            dh = open(filename)
+            try:
+                doc = dh.read()
+            finally:
+                dh.close()
+            test = self.parser.get_doctest(
+                doc, globs={'__file__': filename}, name=name,
+                filename=filename, lineno=0)
+            if test.examples:
+                #print 'FileCase:',test.examples  # dbg
+                yield DocFileCase(test)
+            else:
+                yield False # no tests to load
+
+
+    def wantFile(self,filename):
+        """Return whether the given filename should be scanned for tests.
+
+        Modified version that accepts extension modules as valid containers for
+        doctests.
+        """
+        #print 'Filename:',filename  # dbg
+
+        if is_extension_module(filename):
+            return True
+        else:
+            return doctests.Doctest.wantFile(self,filename)
+
+    # NOTE: the method below is a *copy* of the one in the nose doctests
+    # plugin, but we have to replicate it here in order to have it resolve the
+    # DocTestCase (last line) to our local copy, since the nose plugin doesn't
+    # provide a public hook for what TestCase class to use.  The alternative
+    # would be to monkeypatch doctest in the stdlib, but that's ugly and
+    # brittle, since a change in plugin load order can break it.  So for now,
+    # we just paste this in here, inelegant as this may be.
+
+    def loadTestsFromModule(self, module):
+        #print 'lTM',module  # dbg
+
+        if not self.matches(module.__name__):
+            log.debug("Doctest doesn't want module %s", module)
+            return
+        tests = self.finder.find(module)
+        if not tests:
+            return
+        tests.sort()
+        module_file = module.__file__
+        if module_file[-4:] in ('.pyc', '.pyo'):
+            module_file = module_file[:-1]
+        for test in tests:
+            if not test.examples:
+                continue
+            if not test.filename:
+                test.filename = module_file
+            yield DocTestCase(test)
+
+class IPythonDoctest(ExtensionDoctest):
+    """Nose Plugin that supports doctests in extension modules.
+    """
+    name = 'ipdoctest'   # call nosetests with --with-ipdoctest
+    enabled = True
+
+    def configure(self, options, config):
+
+        Plugin.configure(self, options, config)
+        self.doctest_tests = options.doctest_tests
+        self.extension = tolist(options.doctestExtension)
+        self.parser = IPDocTestParser()
+        #self.finder = DocTestFinder(parser=IPDocTestParser())
+        self.finder = DocTestFinder(parser=self.parser)

IPython/testing/plugin/iptest.py

@@ -0,0 +1,18 b''
+#!/usr/bin/env python
+"""Nose-based test runner.
+"""
+
+from nose.core import main
+from nose.plugins.builtin import plugins
+from nose.plugins.doctests import Doctest
+
+import ipdoctest
+from ipdoctest import IPDocTestRunner
+
+if __name__ == '__main__':
+    print 'WARNING: this code is incomplete!'
+    print
+
+    pp = [x() for x in plugins]  # activate all builtin plugins first
+    main(testRunner=IPDocTestRunner(),
+         plugins=pp+[ipdoctest.IPythonDoctest(),Doctest()])

IPython/testing/plugin/setup.py

@@ -0,0 +1,18 b''
+#!/usr/bin/env python
+"""A Nose plugin to support IPython doctests.
+"""
+
+from setuptools import setup
+
+setup(name='IPython doctest plugin',
+      version='0.1',
+      author='The IPython Team',
+      description = 'Nose plugin to load IPython-extended doctests',
+      license = 'LGPL',
+      py_modules = ['ipdoctest'],
+      entry_points = {
+        'nose.plugins.0.10': ['ipdoctest = ipdoctest:IPythonDoctest',
+                              'extdoctest = ipdoctest:ExtensionDoctest',
+                              ],
+        },
+      )

IPython/testing/plugin/test_combo.txt

@@ -0,0 +1,36 b''
+=======================
+ Combo testing example
+=======================
+
+This is a simple example that mixes ipython doctests::
+
+    In [1]: import code
+
+    In [2]: 2**12
+    Out[2]: 4096
+
+with command-line example information that does *not* get executed::
+
+    $ mpirun -n 4 ipengine --controller-port=10000 --controller-ip=host0
+
+and with literal examples of Python source code::
+
+    controller = dict(host='myhost',
+		      engine_port=None, # default is 10105
+		      control_port=None,
+		      )
+
+    # keys are hostnames, values are the number of engine on that host
+    engines = dict(node1=2,
+		   node2=2,
+		   node3=2,
+		   node3=2,
+		   )
+
+    # Force failure to detect that this test is being run.
+    1/0
+
+These source code examples are executed but no output is compared at all.  An
+error or failure is reported only if an exception is raised.
+
+NOTE: the execution of pure python blocks is not yet working!

IPython/testing/plugin/test_example.txt

@@ -0,0 +1,24 b''
+=====================================
+ Tests in example form - pure python
+=====================================
+
+This file contains doctest examples embedded as code blocks, using normal
+Python prompts.  See the accompanying file for similar examples using IPython
+prompts (you can't mix both types within one file). The following will be run
+as a test::
+
+    >>> 1+1
+    2
+    >>> print "hello"
+    hello
+
+More than one example works::
+
+    >>> s="Hello World"
+
+    >>> s.upper()
+    'HELLO WORLD'
+
+but you should note that the *entire* test file is considered to be a single
+test.  Individual code blocks that fail are printed separately as ``example
+failures``, but the whole file is still counted and reported as one test.

IPython/testing/plugin/test_exampleip.txt

@@ -0,0 +1,30 b''
+=================================
+ Tests in example form - IPython
+=================================
+
+You can write text files with examples that use IPython prompts (as long as you
+use the nose ipython doctest plugin), but you can not mix and match prompt
+styles in a single file.  That is, you either use all ``>>>`` prompts or all
+IPython-style prompts.  Your test suite *can* have both types, you just need to
+put each type of example in a separate. Using IPython prompts, you can paste
+directly from your session::
+
+    In [5]: s="Hello World"
+
+    In [6]: s.upper()
+    Out[6]: 'HELLO WORLD'
+
+Another example::
+
+    In [8]: 1+3
+    Out[8]: 4
+
+Just like in IPython docstrings, you can use all IPython syntax and features::
+
+    In [9]: !echo "hello"
+    hello
+
+    In [10]: a='hi'
+
+    In [11]: !echo $a
+    hi

docs/examples/kernel/asyncmultiengine1.py

@@ -0,0 +1,23 b''
+#!/usr/bin/env python
+# encoding: utf-8
+
+# A super simple example showing how to use all of this in a fully
+# asynchronous manner.  The TaskClient also works in this mode.
+
+from twisted.internet import reactor, defer
+from IPython.kernel import asyncclient
+
+def printer(r):
+    print r
+    return r
+
+def submit(client):
+    d = client.push(dict(a=5, b='asdf', c=[1,2,3]),targets=0,block=True)
+    d.addCallback(lambda _: client.pull(('a','b','c'),targets=0,block=True))
+    d.addBoth(printer)
+    d.addCallback(lambda _: reactor.stop())
+
+d = asyncclient.get_multiengine_client()
+d.addCallback(submit)
+
+reactor.run()
  No newline at end of file

docs/examples/kernel/asynctask1.py

@@ -0,0 +1,32 b''
+#!/usr/bin/env python
+# encoding: utf-8
+
+# This example shows how the AsynTaskClient can be used
+# This example is currently broken
+
+from twisted.internet import reactor, defer
+from IPython.kernel import asyncclient
+
+mec = asyncclient.AsyncMultiEngineClient(('localhost', 10105))
+tc = asyncclient.AsyncTaskClient(('localhost',10113))
+
+cmd1 = """\
+a = 5
+b = 10*d
+c = a*b*d
+"""
+
+t1 = asyncclient.Task(cmd1, clear_before=False, clear_after=True, pull=['a','b','c'])
+
+d = mec.push(dict(d=30))
+
+def raise_and_print(tr):
+    tr.raiseException()
+    print "a, b: ", tr.ns.a, tr.ns.b
+    return tr
+
+d.addCallback(lambda _: tc.run(t1))
+d.addCallback(lambda tid: tc.get_task_result(tid,block=True))
+d.addCallback(raise_and_print)
+d.addCallback(lambda _: reactor.stop())
+reactor.run()

IPython/Magic.py

@@ -2728,8 +2728,7 b' Defaulting color scheme to \'NoColor\'"""'
                 os.chdir(os.path.expanduser(ps))
                 if self.shell.rc.term_title:
                     #print 'set term title:',self.shell.rc.term_title  # dbg
-                    ttitle = 'IPy ' + abbrev_cwd()
-                    platutils.set_term_title(ttitle)
+                    platutils.set_term_title('IPy ' + abbrev_cwd())
             except OSError:
                 print sys.exc_info()[1]
             else:
@@ -3195,7 +3194,7 b' Defaulting color scheme to \'NoColor\'"""'
             exec b in self.user_ns            
             self.user_ns['pasted_block'] = b
         else:
-            self.user_ns[par] = block
+            self.user_ns[par] = SList(block.splitlines())
             print "Block assigned to '%s'" % par
             
     def magic_quickref(self,arg):

IPython/Release.py

@@ -22,15 +22,19 b" name = 'ipython'"
 # because bdist_rpm does not accept dashes (an RPM) convention, and
 # bdist_deb does not accept underscores (a Debian convention).
 
-revision = '1016'
+development = True    # change this to False to do a release
+version_base = '0.9.0'
 branch = 'ipython'
+revision = '1016'
 
+if development:
 if branch == 'ipython':
-    version = '0.9.0.bzr.r' + revision
+        version = '%s.bzr.r%s' % (version_base, revision)
+    else:
+        version = '%s.bzr.r%s.%s' % (version_base, revision, branch)
 else:
-    version = '0.9.0.bzr.r%s.%s'  % (revision,branch)
+    version = version_base
 
-# version = '0.8.4'
 
 description = "Tools for interactive development in Python."
 

IPython/external/mglob.py

@@ -53,7 +53,7 b' globsyntax = """\\'
          - readme*, exclude files ending with .bak
      !.svn/ !.hg/ !*_Data/ rec:.
          - Skip .svn, .hg, foo_Data dirs (and their subdirs) in recurse.
-           Trailing / is the key, \ does not work!
+           Trailing / is the key, \ does not work! Use !.*/ for all hidden.
      dir:foo                       
          - the directory foo if it exists (not files in foo)
      dir:*                         
@@ -63,13 +63,16 b' globsyntax = """\\'
            foo.py is *not* included twice.
      @filelist.txt
          - All files listed in 'filelist.txt' file, on separate lines.
+     "cont:class \wak:" rec:*.py
+         - Match files containing regexp. Applies to subsequent files.
+           note quotes because of whitespace.
  """
 
 
 __version__ = "0.2"
 
 
-import os,glob,fnmatch,sys
+import os,glob,fnmatch,sys,re
 from sets import Set as set
 
                 
@@ -84,21 +87,34 b' def expand(flist,exp_dirs = False):'
     
     """
     if isinstance(flist, basestring):
-        flist = flist.split()
+        import shlex
+        flist = shlex.split(flist)
     done_set = set()
     denied_set = set()
+    cont_set = set()
+    cur_rejected_dirs = set()
 
     def recfind(p, pats = ["*"]):
-        denied_dirs = ["*" + d+"*" for d in denied_set if d.endswith("/")]
-        #print "de", denied_dirs
+        denied_dirs = [os.path.dirname(d) for d in denied_set if d.endswith("/")]
         for (dp,dnames,fnames) in os.walk(p):
             # see if we should ignore the whole directory
             dp_norm = dp.replace("\\","/") + "/"
             deny = False
+            # do not traverse under already rejected dirs
+            for d in cur_rejected_dirs:
+                if dp.startswith(d):
+                    deny = True
+                    break
+            if deny:
+                continue
+            
+
             #print "dp",dp
+            bname = os.path.basename(dp)
             for deny_pat in denied_dirs:
-                if fnmatch.fnmatch( dp_norm, deny_pat):
+                if fnmatch.fnmatch( bname, deny_pat):
                     deny = True
+                    cur_rejected_dirs.add(dp)
                     break
             if deny:
                 continue
@@ -124,6 +140,17 b' def expand(flist,exp_dirs = False):'
                 if fnmatch.fnmatch(os.path.basename(p), deny_pat):
                     deny = True
                     break
+            if cont_set:
+                try:
+                    cont = open(p).read()
+                except IOError:
+                    # deny
+                    continue
+                for pat in cont_set:
+                    if not re.search(pat,cont, re.IGNORECASE):
+                        deny = True
+                        break
+                    
             if not deny:
                 yield it
         return
@@ -158,7 +185,8 b' def expand(flist,exp_dirs = False):'
         # glob only dirs
         elif ent.lower().startswith('dir:'):
             res.extend(once_filter(filter(os.path.isdir,glob.glob(ent[4:]))))
-            
+        elif ent.lower().startswith('cont:'):
+            cont_set.add(ent[5:])
         # get all files in the specified dir
         elif os.path.isdir(ent) and exp_dirs:
             res.extend(once_filter(filter(os.path.isfile,glob.glob(ent + os.sep+"*"))))

IPython/genutils.py

@@ -1137,14 +1137,41 b' class SList(list):'
                 res.append(" ".join(lineparts))
             
         return res        
+    def sort(self,field= None,  nums = False):
+        """ sort by specified fields (see fields())
                 
+        Example::
+            a.sort(1, nums = True)
             
+        Sorts a by second field, in numerical order (so that 21 > 3)
+        
+        """ 
+        
+        #decorate, sort, undecorate
+        if field is not None:
+            dsu = [[SList([line]).fields(field),  line] for line in self]
+        else:
+            dsu = [[line,  line] for line in self]
+        if nums:
+            for i in range(len(dsu)):
+                numstr = "".join([ch for ch in dsu[i][0] if ch.isdigit()])
+                try:
+                    n = int(numstr)
+                except ValueError:
+                    n = 0;
+                dsu[i][0] = n
             
         
+        dsu.sort()
+        return SList([t[1] for t in dsu])
 
 def print_slist(arg):
     """ Prettier (non-repr-like) and more informative printer for SList """
-    print "SList (.p, .n, .l, .s, .grep(), .fields() available). Value:"
+    print "SList (.p, .n, .l, .s, .grep(), .fields(), sort() available):"
+    if hasattr(arg,  'hideonce') and arg.hideonce:
+        arg.hideonce = False
+        return
+        
     nlprint(arg)
     
 print_slist = result_display.when_type(SList)(print_slist)

IPython/kernel/asyncclient.py

@@ -27,7 +27,7 b' from IPython.kernel import codeutil'
 from IPython.kernel.clientconnector import ClientConnector
 
 # Other things that the user will need
-from IPython.kernel.task import Task
+from IPython.kernel.task import MapTask, StringTask
 from IPython.kernel.error import CompositeError
 
 #-------------------------------------------------------------------------------

IPython/kernel/client.py

@@ -44,7 +44,7 b' from IPython.kernel import codeutil'
 import IPython.kernel.magic
 
 # Other things that the user will need
-from IPython.kernel.task import Task
+from IPython.kernel.task import MapTask, StringTask
 from IPython.kernel.error import CompositeError
 
 #-------------------------------------------------------------------------------

IPython/kernel/contexts.py

@@ -144,35 +144,37 b' class RemoteMultiEngine(RemoteContextBase):'
 
 
 # XXX - Temporary hackish testing, we'll move this into proper tests right
-# away
-
-if __name__ == '__main__':
-
-    # XXX - for now, we need a running cluster to be started separately.  The
-    # daemon work is almost finished, and will make much of this unnecessary.
-    from IPython.kernel import client
-    mec = client.MultiEngineClient(('127.0.0.1',10105))
-
-    try:
-        mec.get_ids()
-    except ConnectionRefusedError:
-        import os, time
-        os.system('ipcluster -n 2 &')
-        time.sleep(2)
-        mec = client.MultiEngineClient(('127.0.0.1',10105))
-
-    mec.block = False
-
-    import itertools
-    c = itertools.count()
-
-    parallel = RemoteMultiEngine(mec)
-
-    with parallel as pr:
-        # A comment
-        remote()  # this means the code below only runs remotely
-        print 'Hello remote world'
-        x = 3.14
-        # Comments are OK
-        # Even misindented.
-        y = x+1
+# away.  This has been commented out as it doesn't run under Python 2.4
+# because of the usage of the with statement below.  We need to protect
+# such things with a try:except.
+
+# if __name__ == '__main__':
+# 
+#     # XXX - for now, we need a running cluster to be started separately.  The
+#     # daemon work is almost finished, and will make much of this unnecessary.
+#     from IPython.kernel import client
+#     mec = client.MultiEngineClient(('127.0.0.1',10105))
+# 
+#     try:
+#         mec.get_ids()
+#     except ConnectionRefusedError:
+#         import os, time
+#         os.system('ipcluster -n 2 &')
+#         time.sleep(2)
+#         mec = client.MultiEngineClient(('127.0.0.1',10105))
+# 
+#     mec.block = False
+# 
+#     import itertools
+#     c = itertools.count()
+# 
+#     parallel = RemoteMultiEngine(mec)
+# 
+#     with parallel as pr:
+#         # A comment
+#         remote()  # this means the code below only runs remotely
+#         print 'Hello remote world'
+#         x = 3.14
+#         # Comments are OK
+#         # Even misindented.
+#         y = x+1

IPython/kernel/magic.py

@@ -79,7 +79,7 b" def magic_px(self,parameter_s=''):"
     except AttributeError:
         print NO_ACTIVE_CONTROLLER
     else:
-        print "Executing command on Controller"
+        print "Parallel execution on engines: %s" % activeController.targets
         result = activeController.execute(parameter_s)
         return result
 

IPython/kernel/map.py

@@ -115,7 +115,7 b' class RoundRobinMap(Map):'
         #    result.append(concat[i:totalLength:maxPartitionLength])
         return self.concatenate(listOfPartitions)
         
-styles = {'basic':Map}
+dists = {'b':Map}
 
     
     

IPython/kernel/multiengine.py

@@ -653,67 +653,55 b' components.registerAdapter(SynchronousMultiEngine, IMultiEngine, ISynchronousMul'
 class IMultiEngineCoordinator(Interface):
     """Methods that work on multiple engines explicitly."""
        
-    def scatter(key, seq, style='basic', flatten=False, targets='all'):
-        """Partition and distribute a sequence to targets.
+    def scatter(key, seq, dist='b', flatten=False, targets='all'):
+        """Partition and distribute a sequence to targets."""
         
-        :Parameters:
-            key : str
-                The variable name to call the scattered sequence.
-            seq : list, tuple, array
-                The sequence to scatter.  The type should be preserved.
-            style : string
-                A specification of how the sequence is partitioned.  Currently 
-                only 'basic' is implemented.
-            flatten : boolean
-                Should single element sequences be converted to scalars.
-        """
+    def gather(key, dist='b', targets='all'):
+        """Gather object key from targets."""
         
-    def gather(key, style='basic', targets='all'):
-        """Gather object key from targets.
-    
-        :Parameters:
-            key : string
-                The name of a sequence on the targets to gather.
-            style : string
-                A specification of how the sequence is partitioned.  Currently 
-                only 'basic' is implemented.                
+    def raw_map(func, seqs, dist='b', targets='all'):
         """
+        A parallelized version of Python's builtin `map` function.
     
-    def map(func, seq, style='basic', targets='all'):
-        """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.
         
-        This function implements the following pattern:
+        The equivalence is:
         
-        1. The sequence seq is scattered to the given targets.
-        2. map(functionSource, seq) is called on each engine.
-        3. The resulting sequences are gathered back to the local machine.
+        raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
                 
-        :Parameters:
-            targets : int, list or 'all'
-                The engine ids the action will apply to.  Call `get_ids` to see
-                a list of currently available engines.
-            func : str, function
-                An actual function object or a Python string that names a 
-                callable defined on the engines.
-            seq : list, tuple or numpy array
-                The local sequence to be scattered.
-            style : str
-                Only 'basic' is supported for now.
-                
-        :Returns: A list of len(seq) with functionSource called on each element
-        of seq.
-                
-        Example
-        =======
-        
-        >>> rc.mapAll('lambda x: x*x', range(10000))
-        [0,2,4,9,25,36,...]
+        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."""
-    pass
+
+    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`.
+        """
 
 
 #-------------------------------------------------------------------------------
@@ -722,46 +710,31 b' class ISynchronousMultiEngineCoordinator(IMultiEngineCoordinator):'
 
 class IMultiEngineExtras(Interface):
     
-    def zip_pull(targets, *keys):
-        """Pull, but return results in a different format from `pull`.
+    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.
-        
-        :Parameters:
-            targets : int, list or 'all'
-                The engine ids the action will apply to.  Call `get_ids` to see
-                a list of currently available engines.
-            keys: list or tuple of str
-                A list of variable names as string of the Python objects to be pulled
-                back to the client.
-        
-        :Returns: A list of pulled Python objects for each target.
         """
     
     def run(targets, fname):
-        """Run a .py file on targets.
-        
-        :Parameters:
-            targets : int, list or 'all'
-                The engine ids the action will apply to.  Call `get_ids` to see
-                a list of currently available engines.
-            fname : str
-                The filename of a .py file on the local system to be sent to and run
-                on the engines.
-            block : boolean
-                Should I block or not.  If block=True, wait for the action to
-                complete and return the result.  If block=False, return a
-                `PendingResult` object that can be used to later get the
-                result.  If block is not specified, the block attribute 
-                will be used instead. 
-        """
+        """Run a .py file on targets."""
 
 
 class ISynchronousMultiEngineExtras(IMultiEngineExtras):
-    pass
+    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

IPython/kernel/multiengineclient.py

@@ -31,6 +31,11 b' 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,
@@ -186,6 +191,10 b' class ResultList(list):'
     
     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
@@ -295,34 +304,6 b' class InteractiveMultiEngineClient(object):'
         """Return the number of available engines."""
         return len(self.get_ids())
         
-    def parallelize(self, func, targets=None, block=None):
-        """Build a `ParallelFunction` object for functionName on engines.
-        
-        The returned object will implement a parallel version of functionName
-        that takes a local sequence as its only argument and calls (in 
-        parallel) functionName on each element of that sequence.  The
-        `ParallelFunction` object has a `targets` attribute that controls
-        which engines the function is run on.
-        
-        :Parameters:
-            targets : int, list or 'all'
-                The engine ids the action will apply to.  Call `get_ids` to see
-                a list of currently available engines.
-            functionName : str
-                A Python string that names a callable defined on the engines.
-        
-        :Returns:  A `ParallelFunction` object.                
-        
-        Examples
-        ========
-        
-        >>> psin = rc.parallelize('all','lambda x:sin(x)')
-        >>> psin(range(10000))
-        [0,2,4,9,25,36,...]
-        """
-        targets, block = self._findTargetsAndBlock(targets, block)
-        return ParallelFunction(func, self, targets, block)
-
     #---------------------------------------------------------------------------
     # Make this a context manager for with
     #---------------------------------------------------------------------------
@@ -422,7 +403,11 b' class FullBlockingMultiEngineClient(InteractiveMultiEngineClient):'
     engine, run code on it, etc.
     """
     
-    implements(IFullBlockingMultiEngineClient)
+    implements(
+        IFullBlockingMultiEngineClient,
+        IMultiEngineMapperFactory,
+        IMapper
+    )
     
     def __init__(self, smultiengine):
         self.smultiengine = smultiengine
@@ -779,29 +764,100 b' class FullBlockingMultiEngineClient(InteractiveMultiEngineClient):'
     # IMultiEngineCoordinator
     #---------------------------------------------------------------------------
              
-    def scatter(self, key, seq, style='basic', flatten=False, targets=None, block=None):
+    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, 
-            style, flatten, targets=targets, block=block)
+            dist, flatten, targets=targets, block=block)
     
-    def gather(self, key, style='basic', targets=None, block=None):
+    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, style, 
+        return self._blockFromThread(self.smultiengine.gather, key, dist, 
             targets=targets, block=block)
     
-    def map(self, func, seq, style='basic', targets=None, block=None):
+    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 parallelized version of Python's builtin map
+        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)
-        return self._blockFromThread(self.smultiengine.map, func, seq, 
-            style, targets=targets, block=block)
+        mapper = self.mapper(dist, targets, block)
+        pf = ParallelFunction(mapper)
+        return pf
     
     #---------------------------------------------------------------------------
     # IMultiEngineExtras

IPython/kernel/multienginefc.py

@@ -29,6 +29,12 b' from foolscap import Referenceable'
 from IPython.kernel import error 
 from IPython.kernel.util import printer
 from IPython.kernel import map as Map
+from IPython.kernel.parallelfunction import ParallelFunction
+from IPython.kernel.mapper import (
+    MultiEngineMapper, 
+    IMultiEngineMapperFactory,
+    IMapper
+)
 from IPython.kernel.twistedutil import gatherBoth
 from IPython.kernel.multiengine import (MultiEngine,
     IMultiEngine,
@@ -280,7 +286,12 b' components.registerAdapter(FCSynchronousMultiEngineFromMultiEngine,'
 
 class FCFullSynchronousMultiEngineClient(object):
     
-    implements(IFullSynchronousMultiEngine, IBlockingClientAdaptor)
+    implements(
+        IFullSynchronousMultiEngine, 
+        IBlockingClientAdaptor,
+        IMultiEngineMapperFactory,
+        IMapper
+    )
     
     def __init__(self, remote_reference):
         self.remote_reference = remote_reference
@@ -475,7 +486,7 b' class FCFullSynchronousMultiEngineClient(object):'
         d.addCallback(create_targets)
         return d
     
-    def scatter(self, key, seq, style='basic', flatten=False, targets='all', block=True):
+    def scatter(self, key, seq, dist='b', flatten=False, targets='all', block=True):
         
         # Note: scatter and gather handle pending deferreds locally through self.pdm.
         # This enables us to collect a bunch fo deferred ids and make a secondary 
@@ -483,7 +494,7 b' class FCFullSynchronousMultiEngineClient(object):'
         # difficult to get right though.
         def do_scatter(engines):
             nEngines = len(engines)
-            mapClass = Map.styles[style]
+            mapClass = Map.dists[dist]
             mapObject = mapClass()
             d_list = []
             # Loop through and push to each engine in non-blocking mode.
@@ -541,7 +552,7 b' class FCFullSynchronousMultiEngineClient(object):'
         d.addCallback(do_scatter)
         return d
 
-    def gather(self, key, style='basic', targets='all', block=True):
+    def gather(self, key, dist='b', targets='all', block=True):
         
         # Note: scatter and gather handle pending deferreds locally through self.pdm.
         # This enables us to collect a bunch fo deferred ids and make a secondary 
@@ -549,7 +560,7 b' class FCFullSynchronousMultiEngineClient(object):'
         # difficult to get right though.
         def do_gather(engines):
             nEngines = len(engines)
-            mapClass = Map.styles[style]
+            mapClass = Map.dists[dist]
             mapObject = mapClass()
             d_list = []
             # Loop through and push to each engine in non-blocking mode.
@@ -604,25 +615,103 b' class FCFullSynchronousMultiEngineClient(object):'
         d.addCallback(do_gather)
         return d
 
-    def map(self, func, seq, style='basic', targets='all', block=True):
-        d_list = []
+    def raw_map(self, func, sequences, 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`.
+        """
+        if not isinstance(sequences, (list, tuple)):
+            raise TypeError('sequences must be a list or tuple')
+        max_len = max(len(s) for s in sequences)
+        for s in sequences:
+            if len(s)!=max_len:
+                raise ValueError('all sequences must have equal length')
         if isinstance(func, FunctionType):
             d = self.push_function(dict(_ipython_map_func=func), targets=targets, block=False)
             d.addCallback(lambda did: self.get_pending_deferred(did, True))
-            sourceToRun = '_ipython_map_seq_result = map(_ipython_map_func, _ipython_map_seq)'
+            sourceToRun = '_ipython_map_seq_result = map(_ipython_map_func, *zip(*_ipython_map_seq))'
         elif isinstance(func, str):
             d = defer.succeed(None)
             sourceToRun = \
-                '_ipython_map_seq_result = map(%s, _ipython_map_seq)' % func
+                '_ipython_map_seq_result = map(%s, *zip(*_ipython_map_seq))' % func
         else:
             raise TypeError("func must be a function or str")
         
-        d.addCallback(lambda _: self.scatter('_ipython_map_seq', seq, style, targets=targets))
+        d.addCallback(lambda _: self.scatter('_ipython_map_seq', zip(*sequences), dist, targets=targets))
         d.addCallback(lambda _: self.execute(sourceToRun, targets=targets, block=False))
         d.addCallback(lambda did: self.get_pending_deferred(did, True))
-        d.addCallback(lambda _: self.gather('_ipython_map_seq_result', style, targets=targets, block=block))
+        d.addCallback(lambda _: self.gather('_ipython_map_seq_result', dist, targets=targets, block=block))
         return d
 
+    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=True):
+        """
+        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='all', block=True):
+        """
+        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
+        """
+        mapper = self.mapper(dist, targets, block)
+        pf = ParallelFunction(mapper)
+        return pf
+    
     #---------------------------------------------------------------------------
     # ISynchronousMultiEngineExtras related methods
     #---------------------------------------------------------------------------

IPython/kernel/parallelfunction.py

@@ -16,17 +16,92 b' __docformat__ = "restructuredtext en"'
 #-------------------------------------------------------------------------------
 
 from types import FunctionType
+from zope.interface import Interface, implements
 
-class ParallelFunction:
-    """A function that operates in parallel on sequences."""
-    def __init__(self, func, multiengine, targets, block):
-        """Create a `ParallelFunction`.
+
+class IMultiEngineParallelDecorator(Interface):
+    """A decorator that creates a parallel function."""
+    
+    def parallel(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
+        """
+
+class ITaskParallelDecorator(Interface):
+    """A decorator that creates a parallel function."""
+    
+    def parallel(clear_before=False, clear_after=False, retries=0, 
+        recovery_task=None, depend=None, block=True):
+        """
+        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.
+        
+        See the documentation for `IPython.kernel.task.BaseTask` for 
+        documentation on the arguments to this method.
+        """
+
+class IParallelFunction(Interface):
+    pass
+
+class ParallelFunction(object):
+    """
+    The implementation of a parallel function.
+    
+    A parallel function is similar to Python's map function: 
+    
+    map(func, *sequences) -> pfunc(*sequences)
+    
+    Parallel functions should be created by using the @parallel decorator.
+    """
+    
+    implements(IParallelFunction)
+    
+    def __init__(self, mapper):
+        """
+        Create a parallel function from an `IMapper`.
+        
+        :Parameters:
+            mapper : an `IMapper` implementer.
+                The mapper to use for the parallel function
+        """
+        self.mapper = mapper
+        
+    def __call__(self, func):
+        """
+        Decorate a function to make it run in parallel.
         """
         assert isinstance(func, (str, FunctionType)), "func must be a fuction or str"
         self.func = func
-        self.multiengine = multiengine
-        self.targets = targets
-        self.block = block
+        def call_function(*sequences):
+            return self.mapper.map(self.func, *sequences)
+        return call_function
+
         
  No newline at end of file
-    def __call__(self, sequence):
-        return self.multiengine.map(self.func, sequence, targets=self.targets, block=self.block)
  No newline at end of file

IPython/kernel/scripts/ipcluster.py

@@ -168,8 +168,7 b' def startMsg(control_host,control_port=10105):'
     print 'For interactive use, you can make a MultiEngineClient with:'
     print
     print 'from IPython.kernel import client'
-    print "mec = client.MultiEngineClient((%r,%s))" % \
-          (control_host,control_port)
+    print "mec = client.MultiEngineClient()"
     print
     print 'You can then cleanly stop the cluster from IPython using:'
     print
@@ -191,16 +190,18 b' def clusterLocal(opt,arg):'
         logfile = pjoin(logdir_base,'ipcluster-')
 
     print 'Starting controller:',
-    controller = Popen(['ipcontroller','--logfile',logfile])
+    controller = Popen(['ipcontroller','--logfile',logfile,'-x','-y'])
     print 'Controller PID:',controller.pid
 
     print 'Starting engines:   ',
-    time.sleep(3)
+    time.sleep(5)
 
     englogfile = '%s%s-' % (logfile,controller.pid)
     mpi = opt.mpi
     if mpi: # start with mpi - killing the engines with sigterm will not work if you do this
-        engines = [Popen(['mpirun', '-np', str(opt.n), 'ipengine', '--mpi', mpi, '--logfile',englogfile])]
+        engines = [Popen(['mpirun', '-np', str(opt.n), 'ipengine', '--mpi', 
+            mpi, '--logfile',englogfile])]
+        # engines = [Popen(['mpirun', '-np', str(opt.n), 'ipengine', '--mpi', mpi])]
     else: # do what we would normally do
         engines = [ Popen(['ipengine','--logfile',englogfile])
                     for i in range(opt.n) ]

IPython/kernel/scripts/ipengine.py

@@ -58,12 +58,14 b' def start_engine():'
     kernel_config = kernel_config_manager.get_config_obj()
     core_config = core_config_manager.get_config_obj()
     
+
     # Execute the mpi import statement that needs to call MPI_Init
+    global mpi
     mpikey = kernel_config['mpi']['default']
     mpi_import_statement = kernel_config['mpi'].get(mpikey, None)
     if mpi_import_statement is not None:
         try:
-            exec mpi_import_statement in locals(), globals()
+            exec mpi_import_statement in globals()
         except:
             mpi = None
     else:

IPython/kernel/task.py

@@ -5,116 +5,404 b''
 
 __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 copy, time
-from types import FunctionType as function
+from types import FunctionType
 
 import zope.interface as zi, string
 from twisted.internet import defer, reactor
 from twisted.python import components, log, failure
 
-# from IPython.genutils import time
-
+from IPython.kernel.util import printer
 from IPython.kernel import engineservice as es, error
 from IPython.kernel import controllerservice as cs
 from IPython.kernel.twistedutil import gatherBoth, DeferredList
 
 from IPython.kernel.pickleutil import can,uncan, CannedFunction
 
-def canTask(task):
-    t = copy.copy(task)
-    t.depend = can(t.depend)
-    if t.recovery_task:
-        t.recovery_task = canTask(t.recovery_task)
-    return t
-
-def uncanTask(task):
-    t = copy.copy(task)
-    t.depend = uncan(t.depend)
-    if t.recovery_task and t.recovery_task is not task:
-        t.recovery_task = uncanTask(t.recovery_task)
-    return t
+#-----------------------------------------------------------------------------
+# Definition of the Task objects
+#-----------------------------------------------------------------------------
 
 time_format = '%Y/%m/%d %H:%M:%S'
 
-class Task(object):
-    """Our representation of a task for the `TaskController` interface.
+class ITask(zi.Interface):
+    """
+    This interface provides a generic definition of what constitutes a task.
+    
+    There are two sides to a task.  First a task needs to take input from
+    a user to determine what work is performed by the task.  Second, the
+    task needs to have the logic that knows how to turn that information
+    info specific calls to a worker, through the `IQueuedEngine` interface.
+    
+    Many method in this class get two things passed to them: a Deferred
+    and an IQueuedEngine implementer.  Such methods should register callbacks
+    on the Deferred that use the IQueuedEngine to accomplish something.  See
+    the existing task objects for examples.
+    """
+    
+    zi.Attribute('retries','How many times to retry the task')
+    zi.Attribute('recovery_task','A task to try if the initial one fails')
+    zi.Attribute('taskid','the id of the task')
+    
+    def start_time(result):
+        """
+        Do anything needed to start the timing of the task.
+        
+        Must simply return the result after starting the timers.
+        """
+    
+    def stop_time(result):
+        """
+        Do anything needed to stop the timing of the task.
+        
+        Must simply return the result after stopping the timers.  This 
+        method will usually set attributes that are used by `process_result`
+        in building result of the task.
+        """
+    
+    def pre_task(d, queued_engine):
+        """Do something with the queued_engine before the task is run.
+        
+        This method should simply add callbacks to the input Deferred
+        that do something with the `queued_engine` before the task is run.
+        
+        :Parameters:
+            d : Deferred
+                The deferred that actions should be attached to
+            queued_engine : IQueuedEngine implementer
+                The worker that has been allocated to perform the task
+        """
+    
+    def post_task(d, queued_engine):
+        """Do something with the queued_engine after the task is run.
+        
+        This method should simply add callbacks to the input Deferred
+        that do something with the `queued_engine` before the task is run.
+        
+        :Parameters:
+            d : Deferred
+                The deferred that actions should be attached to
+            queued_engine : IQueuedEngine implementer
+                The worker that has been allocated to perform the task
+        """
+    
+    def submit_task(d, queued_engine):
+        """Submit a task using the `queued_engine` we have been allocated.
+        
+        When a task is ready to run, this method is called.  This method
+        must take the internal information of the task and make suitable
+        calls on the queued_engine to have the actual work done.
+        
+        This method should simply add callbacks to the input Deferred
+        that do something with the `queued_engine` before the task is run.
+        
+        :Parameters:
+            d : Deferred
+                The deferred that actions should be attached to
+            queued_engine : IQueuedEngine implementer
+                The worker that has been allocated to perform the task
+        """
+
+    def process_result(d, result, engine_id):
+        """Take a raw task result.
+        
+        Objects that implement `ITask` can choose how the result of running
+        the task is presented.  This method takes the raw result and 
+        does this logic.  Two example are the `MapTask` which simply returns
+        the raw result or a `Failure` object and the `StringTask` which
+        returns a `TaskResult` object.
+        
+        :Parameters:
+            d : Deferred
+                The deferred that actions should be attached to
+            result : object
+                The raw task result that needs to be wrapped
+            engine_id : int
+                The id of the engine that did the task
+        
+        :Returns:
+            The result, as a tuple of the form: (success, result).
+            Here, success is a boolean indicating if the task
+            succeeded or failed and result is the result.
+        """
 
-    The user should create instances of this class to represent a task that 
-    needs to be done.
+    def check_depend(properties):
+        """Check properties to see if the task should be run.
+        
+        :Parameters:
+            properties : dict
+                A dictionary of properties that an engine has set
+        
+        :Returns:
+            True if the task should be run, False otherwise
+        """
+    
+    def can_task(self):
+        """Serialize (can) any functions in the task for pickling.
+        
+        Subclasses must override this method and make sure that all 
+        functions in the task are canned by calling `can` on the 
+        function.
+        """
+    
+    def uncan_task(self):
+        """Unserialize (uncan) any canned function in the task."""
+
+class BaseTask(object):
+    """
+    Common fuctionality for all objects implementing `ITask`.
+    """
+    
+    zi.implements(ITask)
+    
+    def __init__(self, clear_before=False, clear_after=False, retries=0,
+            recovery_task=None, depend=None):
+        """
+        Make a generic task.
     
     :Parameters:
-        expression : str
-            A str that is valid python code that is the task.
-        pull : str or list of str 
-            The names of objects to be pulled as results.  If not specified, 
-            will return {'result', None}
-        push : dict
-            A dict of objects to be pushed into the engines namespace before
-            execution of the expression.
         clear_before : boolean
-            Should the engine's namespace be cleared before the task is run.
-            Default=False.
+                Should the engines namespace be cleared before the task
+                is run
         clear_after : boolean 
-            Should the engine's namespace be cleared after the task is run.
-            Default=False.
+                Should the engines namespace be clear after the task is run
         retries : int
-            The number of times to resumbit the task if it fails.  Default=0.
-        recovery_task : Task
-            This is the Task to be run when the task has exhausted its retries
-            Default=None.
-        depend : bool function(properties)
-            This is the dependency function for the Task, which determines
-            whether a task can be run on a Worker.  `depend` is called with
-            one argument, the worker's properties dict, and should return
-            True if the worker meets the dependencies or False if it does
-            not.
-            Default=None - run on any worker
-        options : dict
-            Any other keyword options for more elaborate uses of tasks
-    
-    Examples
-    --------
+                The number of times a task should be retries upon failure
+            recovery_task : any task object
+                If a task fails and it has a recovery_task, that is run
+                upon a retry
+            depend : FunctionType
+                A function that is called to test for properties.  This function
+                must take one argument, the properties dict and return a boolean
+        """
+        self.clear_before = clear_before
+        self.clear_after = clear_after
+        self.retries = retries
+        self.recovery_task = recovery_task
+        self.depend = depend
+        self.taskid = None
+    
+    def start_time(self, result):
+        """
+        Start the basic timers.
+        """
+        self.start = time.time()
+        self.start_struct = time.localtime()
+        return result
+    
+    def stop_time(self, result):
+        """
+        Stop the basic timers.
+        """
+        self.stop = time.time()
+        self.stop_struct = time.localtime()
+        self.duration = self.stop - self.start 
+        self.submitted = time.strftime(time_format, self.start_struct)
+        self.completed = time.strftime(time_format)
+        return result
+    
+    def pre_task(self, d, queued_engine):
+        """
+        Clear the engine before running the task if clear_before is set.
+        """
+        if self.clear_before:
+            d.addCallback(lambda r: queued_engine.reset())
+    
+    def post_task(self, d, queued_engine):
+        """
+        Clear the engine after running the task if clear_after is set.
+        """
+        def reseter(result):
+            queued_engine.reset()
+            return result
+        if self.clear_after:
+            d.addBoth(reseter)
+    
+    def submit_task(self, d, queued_engine):
+        raise NotImplementedError('submit_task must be implemented in a subclass')
+    
+    def process_result(self, result, engine_id):
+        """
+        Process a task result.
+        
+        This is the default `process_result` that just returns the raw
+        result or a `Failure`.
+        """
+        if isinstance(result, failure.Failure):
+            return (False, result)
+        else:
+            return (True, result)
+    
+    def check_depend(self, properties):
+        """
+        Calls self.depend(properties) to see if a task should be run.
+        """
+        if self.depend is not None:
+            return self.depend(properties)
+        else:
+            return True
+
+    def can_task(self):
+        self.depend = can(self.depend)
+        if isinstance(self.recovery_task, BaseTask):
+            self.recovery_task.can_task()
+            
+    def uncan_task(self):
+        self.depend = uncan(self.depend)
+        if isinstance(self.recovery_task, BaseTask):
+            self.recovery_task.uncan_task()
+
+class MapTask(BaseTask):
+    """
+    A task that consists of a function and arguments.
+    """
     
-    >>> t = Task('dostuff(args)')
-    >>> t = Task('a=5', pull='a')
-    >>> t = Task('a=5\nb=4', pull=['a','b'])
-    >>> t = Task('os.kill(os.getpid(),9)', retries=100) # this is a bad idea
-        # A dependency case:
-    >>> def hasMPI(props):
-    ...     return props.get('mpi') is not None
-    >>> t = Task('mpi.send(blah,blah)', depend = hasMPI)
+    zi.implements(ITask)
+    
+    def __init__(self, function, args=None, kwargs=None, clear_before=False, 
+            clear_after=False, retries=0, recovery_task=None, depend=None):
+        """
+        Create a task based on a function, args and kwargs.
+        
+        This is a simple type of task that consists of calling:
+        function(*args, **kwargs) and wrapping the result in a `TaskResult`.
+        
+        The return value of the function, or a `Failure` wrapping an 
+        exception is the task result for this type of task.
+        """
+        BaseTask.__init__(self, clear_before, clear_after, retries, 
+            recovery_task, depend)
+        if not isinstance(function, FunctionType):
+            raise TypeError('a task function must be a FunctionType')
+        self.function = function
+        if args is None:
+            self.args = ()
+        else:
+            self.args = args
+        if not isinstance(self.args, (list, tuple)):
+            raise TypeError('a task args must be a list or tuple')
+        if kwargs is None:
+            self.kwargs = {}
+        else:
+            self.kwargs = kwargs
+        if not isinstance(self.kwargs, dict):
+            raise TypeError('a task kwargs must be a dict')
+    
+    def submit_task(self, d, queued_engine):
+        d.addCallback(lambda r: queued_engine.push_function(
+            dict(_ipython_task_function=self.function))
+        )
+        d.addCallback(lambda r: queued_engine.push(
+            dict(_ipython_task_args=self.args,_ipython_task_kwargs=self.kwargs))
+        )
+        d.addCallback(lambda r: queued_engine.execute(
+            '_ipython_task_result = _ipython_task_function(*_ipython_task_args,**_ipython_task_kwargs)')
+        )
+        d.addCallback(lambda r: queued_engine.pull('_ipython_task_result'))
+    
+    def can_task(self):
+        self.function = can(self.function)
+        BaseTask.can_task(self)
+    
+    def uncan_task(self):
+        self.function = uncan(self.function)
+        BaseTask.uncan_task(self)
+
+
+class StringTask(BaseTask):
+    """
+    A task that consists of a string of Python code to run.
     """
     
     def __init__(self, expression, pull=None, push=None,
             clear_before=False, clear_after=False, retries=0, 
-            recovery_task=None, depend=None, **options):
+            recovery_task=None, depend=None):
+        """
+        Create a task based on a Python expression and variables
+        
+        This type of task lets you push a set of variables to the engines
+        namespace, run a Python string in that namespace and then bring back
+        a different set of Python variables as the result.
+        
+        Because this type of task can return many results (through the
+        `pull` keyword argument) it returns a special `TaskResult` object
+        that wraps the pulled variables, statistics about the run and
+        any exceptions raised.
+        """
+        if not isinstance(expression, str):
+            raise TypeError('a task expression must be a string')
         self.expression = expression
-        if isinstance(pull, str):
-            self.pull = [pull]
-        else:
+        
+        if pull==None:
+            self.pull = ()
+        elif isinstance(pull, str):
+            self.pull = (pull,)
+        elif isinstance(pull, (list, tuple)):
             self.pull = pull
+        else:
+            raise TypeError('pull must be str or a sequence of strs')
+        
+        if push==None:
+            self.push = {}
+        elif isinstance(push, dict):
         self.push = push
-        self.clear_before = clear_before
-        self.clear_after = clear_after
-        self.retries=retries
-        self.recovery_task = recovery_task
-        self.depend = depend
-        self.options = options
-        self.taskid = None
+        else:
+            raise TypeError('push must be a dict')
+        
+        BaseTask.__init__(self, clear_before, clear_after, retries, 
+            recovery_task, depend)
 
-class ResultNS:
-    """The result namespace object for use in TaskResult objects as tr.ns.
+    def submit_task(self, d, queued_engine):
+        if self.push is not None:
+            d.addCallback(lambda r: queued_engine.push(self.push))
+        
+        d.addCallback(lambda r: queued_engine.execute(self.expression))
+        
+        if self.pull is not None:
+            d.addCallback(lambda r: queued_engine.pull(self.pull))
+        else:
+            d.addCallback(lambda r: None)    
+    
+    def process_result(self, result, engine_id):
+        if isinstance(result, failure.Failure):
+            tr = TaskResult(result, engine_id)
+        else:
+            if self.pull is None:
+                resultDict = {}
+            elif len(self.pull) == 1:
+                resultDict = {self.pull[0]:result}
+            else:
+                resultDict = dict(zip(self.pull, result))
+            tr = TaskResult(resultDict, engine_id)
+        # Assign task attributes
+        tr.submitted = self.submitted
+        tr.completed = self.completed
+        tr.duration = self.duration
+        if hasattr(self,'taskid'):
+            tr.taskid = self.taskid
+        else:
+            tr.taskid = None
+        if isinstance(result, failure.Failure):
+            return (False, tr)
+        else:
+            return (True, tr)
+
+class ResultNS(object):
+    """
+    A dict like object for holding the results of a task.
+    
+    The result namespace object for use in `TaskResult` objects as tr.ns.
     It builds an object from a dictionary, such that it has attributes
     according to the key,value pairs of the dictionary.
     
@@ -152,7 +440,7 b' class ResultNS:'
 
 class TaskResult(object):
     """
-    An object for returning task results.
+    An object for returning task results for certain types of tasks.
         
     This object encapsulates the results of a task.  On task
     success it will have a keys attribute that will have a list
@@ -162,21 +450,21 b' class TaskResult(object):'
         
     In task failure, keys will be empty, but failure will contain
     the failure object that encapsulates the remote exception.
-    One can also simply call the raiseException() method of 
+    One can also simply call the `raise_exception` method of 
     this class to re-raise any remote exception in the local
     session.
         
-    The TaskResult has a .ns member, which is a property for access
+    The `TaskResult` has a `.ns` member, which is a property for access
     to the results.  If the Task had pull=['a', 'b'], then the 
-    Task Result will have attributes tr.ns.a, tr.ns.b for those values.
-    Accessing tr.ns will raise the remote failure if the task failed.
+    Task Result will have attributes `tr.ns.a`, `tr.ns.b` for those values.
+    Accessing `tr.ns` will raise the remote failure if the task failed.
         
-    The engineid attribute should have the engineid of the engine
-    that ran the task.  But, because engines can come and go in
-    the ipython task system, the engineid may not continue to be
+    The `engineid` attribute should have the `engineid` of the engine
+    that ran the task.  But, because engines can come and go, 
+    the `engineid` may not continue to be
     valid or accurate.
         
-    The taskid attribute simply gives the taskid that the task
+    The `taskid` attribute simply gives the `taskid` that the task
     is tracked under.
     """
     taskid = None
@@ -188,7 +476,7 b' class TaskResult(object):'
             return self._ns
     
     def _setNS(self, v):
-        raise Exception("I am protected!")
+        raise Exception("the ns attribute cannot be changed")
     
     ns = property(_getNS, _setNS)
     
@@ -214,15 +502,19 b' class TaskResult(object):'
     
     def __getitem__(self, key):
         if self.failure is not None:
-            self.raiseException()
+            self.raise_exception()
         return self.results[key]
     
-    def raiseException(self):
+    def raise_exception(self):
         """Re-raise any remote exceptions in the local python session."""
         if self.failure is not None:
             self.failure.raiseException()
 
 
+#-----------------------------------------------------------------------------
+# The controller side of things
+#-----------------------------------------------------------------------------
+
 class IWorker(zi.Interface):
     """The Basic Worker Interface. 
     
@@ -237,12 +529,15 b' class IWorker(zi.Interface):'
         :Parameters:
             task : a `Task` object
         
-        :Returns: `Deferred` to a `TaskResult` object.
+        :Returns: `Deferred` to a tuple of (success, result) where
+            success if a boolean that signifies success or failure
+            and result is the task result.
         """
 
 
 class WorkerFromQueuedEngine(object):
     """Adapt an `IQueuedEngine` to an `IWorker` object"""
+    
     zi.implements(IWorker)
     
     def __init__(self, qe):
@@ -257,52 +552,26 b' class WorkerFromQueuedEngine(object):'
     def run(self, task):
         """Run task in worker's namespace.
         
+        This takes a task and calls methods on the task that actually
+        cause `self.queuedEngine` to do the task.  See the methods of
+        `ITask` for more information about how these methods are called.
+        
         :Parameters:
             task : a `Task` object
         
-        :Returns: `Deferred` to a `TaskResult` object.
+        :Returns: `Deferred` to a tuple of (success, result) where
+            success if a boolean that signifies success or failure
+            and result is the task result.
         """
-        if task.clear_before:
-            d = self.queuedEngine.reset()
-        else:
             d = defer.succeed(None)
-            
-        if task.push is not None:
-            d.addCallback(lambda r: self.queuedEngine.push(task.push))
-        
-        d.addCallback(lambda r: self.queuedEngine.execute(task.expression))
-        
-        if task.pull is not None:
-            d.addCallback(lambda r: self.queuedEngine.pull(task.pull))
-        else:
-            d.addCallback(lambda r: None)
-        
-        def reseter(result):
-            self.queuedEngine.reset()
-            return result
-            
-        if task.clear_after:
-            d.addBoth(reseter)
-        
-        return d.addBoth(self._zipResults, task.pull, time.time(), time.localtime())
-    
-    def _zipResults(self, result, names, start, start_struct):
-        """Callback for construting the TaskResult object."""
-        if isinstance(result, failure.Failure):
-            tr = TaskResult(result, self.queuedEngine.id)
-        else:
-            if names is None:
-                resultDict = {} 
-            elif len(names) == 1:
-                resultDict = {names[0]:result}
-            else:
-                resultDict = dict(zip(names, result))
-            tr = TaskResult(resultDict, self.queuedEngine.id)
-        # the time info
-        tr.submitted = time.strftime(time_format, start_struct)
-        tr.completed = time.strftime(time_format)
-        tr.duration = time.time()-start
-        return tr
+        d.addCallback(task.start_time)
+        task.pre_task(d, self.queuedEngine)
+        task.submit_task(d, self.queuedEngine)
+        task.post_task(d, self.queuedEngine)
+        d.addBoth(task.stop_time)
+        d.addBoth(task.process_result, self.queuedEngine.id)
+        # At this point, there will be (success, result) coming down the line
+        return d
     
 
 components.registerAdapter(WorkerFromQueuedEngine, es.IEngineQueued, IWorker)
@@ -319,14 +588,14 b' class IScheduler(zi.Interface):'
         """Add a task to the queue of the Scheduler.
         
         :Parameters:
-            task : a `Task` object
+            task : an `ITask` implementer
                 The task to be queued.
             flags : dict
                 General keywords for more sophisticated scheduling
         """
     
     def pop_task(id=None):
-        """Pops a Task object.
+        """Pops a task object from the queue.
         
         This gets the next task to be run.  If no `id` is requested, the highest priority
         task is returned.
@@ -336,7 +605,7 b' class IScheduler(zi.Interface):'
                 The id of the task to be popped.  The default (None) is to return 
                 the highest priority task.
                 
-        :Returns: a `Task` object
+        :Returns: an `ITask` implementer
         
         :Exceptions:
             IndexError : raised if no taskid in queue
@@ -346,8 +615,9 b' class IScheduler(zi.Interface):'
         """Add a worker to the worker queue.
         
         :Parameters:
-            worker : an IWorker implementing object
-            flags : General keywords for more sophisticated scheduling
+            worker : an `IWorker` implementer
+            flags : dict
+                General keywords for more sophisticated scheduling
         """
     
     def pop_worker(id=None):
@@ -370,15 +640,15 b' class IScheduler(zi.Interface):'
         """Returns True if there is something to do, False otherwise"""
     
     def schedule():
-        """Returns a tuple of the worker and task pair for the next
-        task to be run.
-        """
+        """Returns (worker,task) pair for the next task to be run."""
     
 
 class FIFOScheduler(object):
-    """A basic First-In-First-Out (Queue) Scheduler.
-    This is the default Scheduler for the TaskController.
-    See the docstrings for IScheduler for interface details.
+    """
+    A basic First-In-First-Out (Queue) Scheduler.
+    
+    This is the default Scheduler for the `TaskController`.
+    See the docstrings for `IScheduler` for interface details.
     """
     
     zi.implements(IScheduler)
@@ -435,7 +705,9 b' class FIFOScheduler(object):'
         for t in self.tasks:
             for w in self.workers:
                 try:# do not allow exceptions to break this
-                    cando = t.depend is None or t.depend(w.properties)
+                    # Allow the task to check itself using its
+                    # check_depend method.
+                    cando = t.check_depend(w.properties)
                 except:
                     cando = False
                 if cando:
@@ -445,9 +717,12 b' class FIFOScheduler(object):'
 
 
 class LIFOScheduler(FIFOScheduler):
-    """A Last-In-First-Out (Stack) Scheduler.  This scheduler should naively
-    reward fast engines by giving them more jobs.  This risks starvation, but
-    only in cases with low load, where starvation does not really matter.
+    """
+    A Last-In-First-Out (Stack) Scheduler.
+    
+    This scheduler should naively reward fast engines by giving 
+    them more jobs.  This risks starvation, but only in cases with 
+    low load, where starvation does not really matter.
     """
     
     def add_task(self, task, **flags):
@@ -462,13 +737,15 b' class LIFOScheduler(FIFOScheduler):'
     
 
 class ITaskController(cs.IControllerBase):
-    """The Task based interface to a `ControllerService` object
+    """
+    The Task based interface to a `ControllerService` object
     
     This adapts a `ControllerService` to the ITaskController interface.
     """
     
     def run(task):
-        """Run a task.
+        """
+        Run a task.
         
         :Parameters:
             task : an IPython `Task` object
@@ -477,13 +754,14 b' class ITaskController(cs.IControllerBase):'
         """
     
     def get_task_result(taskid, block=False):
-        """Get the result of a task by its ID.
+        """
+        Get the result of a task by its ID.
         
         :Parameters:
             taskid : int
                 the id of the task whose result is requested
         
-        :Returns: `Deferred` to (taskid, actualResult) if the task is done, and None
+        :Returns: `Deferred` to the task result if the task is done, and None
             if not.
         
         :Exceptions:
@@ -508,23 +786,35 b' class ITaskController(cs.IControllerBase):'
         """
     
     def barrier(taskids):
-        """Block until the list of taskids are completed.
+        """
+        Block until the list of taskids are completed.
         
         Returns None on success.
         """
     
     def spin():
-        """touch the scheduler, to resume scheduling without submitting
-        a task.
+        """
+        Touch the scheduler, to resume scheduling without submitting a task.
         """
     
-    def queue_status(self, verbose=False):
-        """Get a dictionary with the current state of the task queue.
+    def queue_status(verbose=False):
+        """
+        Get a dictionary with the current state of the task queue.
         
         If verbose is True, then return lists of taskids, otherwise, 
         return the number of tasks with each status.
         """
     
+    def clear():
+        """
+        Clear all previously run tasks from the task controller.
+        
+        This is needed because the task controller keep all task results
+        in memory.  This can be a problem is there are many completed
+        tasks.  Users should call this periodically to clean out these
+        cached task results.
+        """
+    
 
 class TaskController(cs.ControllerAdapterBase):
     """The Task based interface to a Controller object.
@@ -561,7 +851,7 b' class TaskController(cs.ControllerAdapterBase):'
     def registerWorker(self, id):
         """Called by controller.register_engine."""
         if self.workers.get(id):
-            raise "We already have one!  This should not happen."
+            raise ValueError("worker with id %s already exists.  This should not happen." % id)
         self.workers[id] = IWorker(self.controller.engines[id])
         self.workers[id].workerid = id
         if not self.pendingTasks.has_key(id):# if not working
@@ -586,21 +876,25 b' class TaskController(cs.ControllerAdapterBase):'
     #---------------------------------------------------------------------------
     
     def run(self, task):
-        """Run a task and return `Deferred` to its taskid."""
+        """
+        Run a task and return `Deferred` to its taskid.
+        """
         task.taskid = self.taskid
         task.start = time.localtime()
         self.taskid += 1
         d = defer.Deferred()
         self.scheduler.add_task(task)
-        # log.msg('Queuing task: %i' % task.taskid)
+        log.msg('Queuing task: %i' % task.taskid)
         
         self.deferredResults[task.taskid] = []
         self.distributeTasks()
         return defer.succeed(task.taskid)
     
     def get_task_result(self, taskid, block=False):
-        """Returns a `Deferred` to a TaskResult tuple or None."""
-        # log.msg("Getting task result: %i" % taskid)
+        """
+        Returns a `Deferred` to the task result, or None.
+        """
+        log.msg("Getting task result: %i" % taskid)
         if self.finishedResults.has_key(taskid):
             tr = self.finishedResults[taskid]
             return defer.succeed(tr)
@@ -615,7 +909,9 b' class TaskController(cs.ControllerAdapterBase):'
             return defer.fail(IndexError("task ID not registered: %r" % taskid))
     
     def abort(self, taskid):
-        """Remove a task from the queue if it has not been run already."""
+        """
+        Remove a task from the queue if it has not been run already.
+        """
         if not isinstance(taskid, int):
             return defer.fail(failure.Failure(TypeError("an integer task id expected: %r" % taskid)))
         try:
@@ -674,8 +970,10 b' class TaskController(cs.ControllerAdapterBase):'
     #---------------------------------------------------------------------------
     
     def _doAbort(self, taskid):
-        """Helper function for aborting a pending task."""
-        # log.msg("Task aborted: %i" % taskid)
+        """
+        Helper function for aborting a pending task.
+        """
+        log.msg("Task aborted: %i" % taskid)
         result = failure.Failure(error.TaskAborted())
         self._finishTask(taskid, result)
         if taskid in self.abortPending:
@@ -683,14 +981,16 b' class TaskController(cs.ControllerAdapterBase):'
     
     def _finishTask(self, taskid, result):
         dlist = self.deferredResults.pop(taskid)
-        result.taskid = taskid   # The TaskResult should save the taskid
+        # result.taskid = taskid   # The TaskResult should save the taskid
         self.finishedResults[taskid] = result
         for d in dlist:
             d.callback(result)
     
     def distributeTasks(self):
-        """Distribute tasks while self.scheduler has things to do."""
-        # log.msg("distributing Tasks")
+        """
+        Distribute tasks while self.scheduler has things to do.
+        """
+        log.msg("distributing Tasks")
         worker, task = self.scheduler.schedule()
         if not worker and not task:
             if self.idleLater and self.idleLater.called:# we are inside failIdle
@@ -705,7 +1005,7 b' class TaskController(cs.ControllerAdapterBase):'
             self.pendingTasks[worker.workerid] = task
             # run/link callbacks
             d = worker.run(task)
-            # log.msg("Running task %i on worker %i" %(task.taskid, worker.workerid))
+            log.msg("Running task %i on worker %i" %(task.taskid, worker.workerid))
             d.addBoth(self.taskCompleted, task.taskid, worker.workerid)
             worker, task = self.scheduler.schedule()
         # check for idle timeout:
@@ -727,14 +1027,15 b' class TaskController(cs.ControllerAdapterBase):'
                 t = self.scheduler.pop_task()
                 msg = "task %i failed to execute due to unmet dependencies"%t.taskid
                 msg += " for %i seconds"%self.timeout
-                # log.msg("Task aborted by timeout: %i" % t.taskid)
+                log.msg("Task aborted by timeout: %i" % t.taskid)
                 f = failure.Failure(error.TaskTimeout(msg))
                 self._finishTask(t.taskid, f)
         self.idleLater = None
                 
     
-    def taskCompleted(self, result, taskid, workerid):
+    def taskCompleted(self, success_and_result, taskid, workerid):
         """This is the err/callback for a completed task."""
+        success, result = success_and_result
         try:
             task = self.pendingTasks.pop(workerid)
         except:
@@ -751,7 +1052,7 b' class TaskController(cs.ControllerAdapterBase):'
             aborted = True
         
         if not aborted:
-            if result.failure is not None and isinstance(result.failure, failure.Failure): # we failed
+            if not success:
                 log.msg("Task %i failed on worker %i"% (taskid, workerid))
                 if task.retries > 0: # resubmit
                     task.retries -= 1
@@ -759,7 +1060,7 b' class TaskController(cs.ControllerAdapterBase):'
                     s = "Resubmitting task %i, %i retries remaining" %(taskid, task.retries)
                     log.msg(s)
                     self.distributeTasks()
-                elif isinstance(task.recovery_task, Task) and \
+                elif isinstance(task.recovery_task, BaseTask) and \
                                     task.recovery_task.retries > -1:
                     # retries = -1 is to prevent infinite recovery_task loop
                     task.retries = -1 
@@ -775,17 +1076,18 b' class TaskController(cs.ControllerAdapterBase):'
                 # it may have died, and not yet been unregistered
                 reactor.callLater(self.failurePenalty, self.readmitWorker, workerid)
             else: # we succeeded
-                # log.msg("Task completed: %i"% taskid)
+                log.msg("Task completed: %i"% taskid)
                 self._finishTask(taskid, result)
                 self.readmitWorker(workerid)
         else:# we aborted the task
-            if result.failure is not None and isinstance(result.failure, failure.Failure): # it failed, penalize worker
+            if not success:
                 reactor.callLater(self.failurePenalty, self.readmitWorker, workerid)
             else:
                 self.readmitWorker(workerid)
     
     def readmitWorker(self, workerid):
-        """Readmit a worker to the scheduler.  
+        """
+        Readmit a worker to the scheduler.  
         
         This is outside `taskCompleted` because of the `failurePenalty` being 
         implemented through `reactor.callLater`.
@@ -795,5 +1097,17 b' class TaskController(cs.ControllerAdapterBase):'
             self.scheduler.add_worker(self.workers[workerid])
             self.distributeTasks()
         
+    def clear(self):
+        """
+        Clear all previously run tasks from the task controller.
+        
+        This is needed because the task controller keep all task results
+        in memory.  This can be a problem is there are many completed
+        tasks.  Users should call this periodically to clean out these
+        cached task results.
+        """
+        self.finishedResults = {}
+        return defer.succeed(None)
+        
     
 components.registerAdapter(TaskController, cs.IControllerBase, ITaskController)

IPython/kernel/taskclient.py

@@ -1,9 +1,8 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.tests.test_taskcontrollerxmlrpc -*-
 
-"""The Generic Task Client object.  
-
-This must be subclassed based on your connection method.
+"""
+A blocking version of the task client.
 """
 
 __docformat__ = "restructuredtext en"
@@ -24,116 +23,97 b' from twisted.python import components, log'
 
 from IPython.kernel.twistedutil import blockingCallFromThread
 from IPython.kernel import task, error
+from IPython.kernel.mapper import (
+    SynchronousTaskMapper,
+    ITaskMapperFactory,
+    IMapper
+)
+from IPython.kernel.parallelfunction import (
+    ParallelFunction, 
+    ITaskParallelDecorator
+)
 
 #-------------------------------------------------------------------------------
-# Connecting Task Client
+# The task client
 #-------------------------------------------------------------------------------
 
-class InteractiveTaskClient(object):
-    
-    def irun(self, *args, **kwargs):
-        """Run a task on the `TaskController`.
-        
-        This method is a shorthand for run(task) and its arguments are simply
-        passed onto a `Task` object:
-        
-        irun(*args, **kwargs) -> run(Task(*args, **kwargs))
-
-        :Parameters:
-            expression : str
-                A str that is valid python code that is the task.
-            pull : str or list of str 
-                The names of objects to be pulled as results.
-            push : dict
-                A dict of objects to be pushed into the engines namespace before
-                execution of the expression.
-            clear_before : boolean
-                Should the engine's namespace be cleared before the task is run.
-                Default=False.
-            clear_after : boolean 
-                Should the engine's namespace be cleared after the task is run.
-                Default=False.
-            retries : int
-                The number of times to resumbit the task if it fails.  Default=0.
-            options : dict
-                Any other keyword options for more elaborate uses of tasks
-            
-        :Returns: A `TaskResult` object.      
-        """
-        block = kwargs.pop('block', False)
-        if len(args) == 1 and isinstance(args[0], task.Task):
-            t = args[0]
-        else:
-            t = task.Task(*args, **kwargs)
-        taskid = self.run(t)
-        print "TaskID = %i"%taskid
-        if block:
-            return self.get_task_result(taskid, block)
-        else:
-            return taskid
-
 class IBlockingTaskClient(Interface):
     """
-    An interface for blocking task clients.
+    A vague interface of the blocking task client
     """
     pass
 
-
-class BlockingTaskClient(InteractiveTaskClient):
+class BlockingTaskClient(object):
     """
-    This class provides a blocking task client.
+    A blocking task client that adapts a non-blocking one.
     """
     
-    implements(IBlockingTaskClient)
+    implements(
+        IBlockingTaskClient, 
+        ITaskMapperFactory,
+        IMapper,
+        ITaskParallelDecorator
+    )
     
     def __init__(self, task_controller):
         self.task_controller = task_controller
         self.block = True
         
-    def run(self, task):
-        """
-        Run a task and return a task id that can be used to get the task result.
+    def run(self, task, block=False):
+        """Run a task on the `TaskController`.
+        
+        See the documentation of the `MapTask` and `StringTask` classes for 
+        details on how to build a task of different types.
         
         :Parameters:
-            task : `Task`
-                The `Task` object to run
+            task : an `ITask` implementer
+        
+        :Returns: The int taskid of the submitted task.  Pass this to 
+            `get_task_result` to get the `TaskResult` object.
         """
-        return blockingCallFromThread(self.task_controller.run, task)
+        tid = blockingCallFromThread(self.task_controller.run, task)
+        if block:
+            return self.get_task_result(tid, block=True)
+        else:
+            return tid
     
     def get_task_result(self, taskid, block=False):
         """
-        Get or poll for a task result.
+        Get a task result by taskid.
         
         :Parameters:
             taskid : int
-                The id of the task whose result to get
+                The taskid of the task to be retrieved.
             block : boolean
-                If True, wait until the task is done and then result the
-                `TaskResult` object.  If False, just poll for the result and
-                return None if the task is not done.
+                Should I block until the task is done?
+        
+        :Returns: A `TaskResult` object that encapsulates the task result.
         """
         return blockingCallFromThread(self.task_controller.get_task_result,
             taskid, block)
     
     def abort(self, taskid):
         """
-        Abort a task by task id if it has not been started.
+        Abort a task by taskid.
+        
+        :Parameters:
+            taskid : int
+                The taskid of the task to be aborted.
         """
         return blockingCallFromThread(self.task_controller.abort, taskid)
     
     def barrier(self, taskids):
-        """
-        Wait for a set of tasks to finish.
+        """Block until a set of tasks are completed.
         
         :Parameters:
-            taskids : list of ints
-                A list of task ids to wait for.
+            taskids : list, tuple
+                A sequence of taskids to block on.
         """
         return blockingCallFromThread(self.task_controller.barrier, taskids)
     
     def spin(self):
         """
-        Cause the scheduler to schedule tasks.
+        Touch the scheduler, to resume scheduling without submitting a task.
         
         This method only needs to be called in unusual situations where the
         scheduler is idle for some reason.
@@ -154,6 +134,45 b' class BlockingTaskClient(InteractiveTaskClient):'
         """
         return blockingCallFromThread(self.task_controller.queue_status, verbose)
 
+    def clear(self):
+        """
+        Clear all previously run tasks from the task controller.
+        
+        This is needed because the task controller keep all task results
+        in memory.  This can be a problem is there are many completed
+        tasks.  Users should call this periodically to clean out these
+        cached task results.
+        """
+        return blockingCallFromThread(self.task_controller.clear)
+    
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is load balanced.
+        """
+        return self.mapper().map(func, *sequences)
+
+    def mapper(self, clear_before=False, clear_after=False, retries=0, 
+                recovery_task=None, depend=None, block=True):
+        """
+        Create an `IMapper` implementer with a given set of arguments.
+        
+        The `IMapper` created using a task controller is load balanced.
+        
+        See the documentation for `IPython.kernel.task.BaseTask` for 
+        documentation on the arguments to this method.
+        """
+        return SynchronousTaskMapper(self, clear_before=clear_before, 
+            clear_after=clear_after, retries=retries, 
+            recovery_task=recovery_task, depend=depend, block=block)
+    
+    def parallel(self, clear_before=False, clear_after=False, retries=0, 
+        recovery_task=None, depend=None, block=True):
+        mapper = self.mapper(clear_before, clear_after, retries,
+            recovery_task, depend, block)
+        pf = ParallelFunction(mapper)
+        return pf
 
 components.registerAdapter(BlockingTaskClient,
             task.ITaskController, IBlockingTaskClient)

IPython/kernel/taskfc.py

@@ -34,6 +34,15 b' from IPython.kernel.clientinterfaces import ('
     IFCClientInterfaceProvider, 
     IBlockingClientAdaptor
 )
+from IPython.kernel.mapper import (
+    TaskMapper,
+    ITaskMapperFactory,
+    IMapper
+)
+from IPython.kernel.parallelfunction import (
+    ParallelFunction, 
+    ITaskParallelDecorator
+)
 
 #-------------------------------------------------------------------------------
 # The Controller side of things
@@ -43,32 +52,38 b' from IPython.kernel.clientinterfaces import ('
 class IFCTaskController(Interface):
     """Foolscap interface to task controller.
         
-    See the documentation of ITaskController for documentation about the methods.
+    See the documentation of `ITaskController` for more information.
     """
-    def remote_run(request, binTask):
+    def remote_run(binTask):
         """"""
     
-    def remote_abort(request, taskid):
+    def remote_abort(taskid):
         """"""
         
-    def remote_get_task_result(request, taskid, block=False):
+    def remote_get_task_result(taskid, block=False):
         """"""
         
-    def remote_barrier(request, taskids):
+    def remote_barrier(taskids):
         """"""
     
-    def remote_spin(request):
+    def remote_spin():
         """"""
     
-    def remote_queue_status(request, verbose):
+    def remote_queue_status(verbose):
+        """"""
+    
+    def remote_clear():
         """"""
 
 
 class FCTaskControllerFromTaskController(Referenceable):
-    """XML-RPC attachmeot for controller.
+    """
+    Adapt a `TaskController` to an `IFCTaskController`
         
-    See IXMLRPCTaskController and ITaskController (and its children) for documentation. 
+    This class is used to expose a `TaskController` over the wire using
+    the Foolscap network protocol.
     """
+    
     implements(IFCTaskController, IFCClientInterfaceProvider)
     
     def __init__(self, taskController):
@@ -92,8 +107,8 b' class FCTaskControllerFromTaskController(Referenceable):'
     
     def remote_run(self, ptask):
         try:
-            ctask = pickle.loads(ptask)
-            task = taskmodule.uncanTask(ctask)
+            task = pickle.loads(ptask)
+            task.uncan_task()
         except:
             d = defer.fail(pickle.UnpickleableError("Could not unmarshal task"))
         else:
@@ -132,6 +147,9 b' class FCTaskControllerFromTaskController(Referenceable):'
         d.addErrback(self.packageFailure)
         return d
     
+    def remote_clear(self):
+        return self.taskController.clear()
+    
     def remote_get_client_name(self):
         return 'IPython.kernel.taskfc.FCTaskClient'
     
@@ -144,13 +162,23 b' components.registerAdapter(FCTaskControllerFromTaskController,'
 #-------------------------------------------------------------------------------
 
 class FCTaskClient(object):
-    """XML-RPC based TaskController client that implements ITaskController.
+    """
+    Client class for Foolscap exposed `TaskController`.
         
-    :Parameters:
-        addr : (ip, port)
-            The ip (str) and port (int) tuple of the `TaskController`.  
+    This class is an adapter that makes a `RemoteReference` to a 
+    `TaskController` look like an actual `ITaskController` on the client side.
+
+    This class also implements `IBlockingClientAdaptor` so that clients can 
+    automatically get a blocking version of this class.
     """
-    implements(taskmodule.ITaskController, IBlockingClientAdaptor)
+    
+    implements(
+        taskmodule.ITaskController, 
+        IBlockingClientAdaptor,
+        ITaskMapperFactory,
+        IMapper,
+        ITaskParallelDecorator
+    )
     
     def __init__(self, remote_reference):
         self.remote_reference = remote_reference
@@ -168,48 +196,26 b' class FCTaskClient(object):'
     def run(self, task):
         """Run a task on the `TaskController`.
         
+        See the documentation of the `MapTask` and `StringTask` classes for 
+        details on how to build a task of different types.
+        
         :Parameters:
-            task : a `Task` object
-        
-        The Task object is created using the following signature:
-        
-        Task(expression, pull=None, push={}, clear_before=False, 
-            clear_after=False, retries=0, **options):)
-        
-        The meaning of the arguments is as follows:
-        
-        :Task Parameters:
-            expression : str
-                A str that is valid python code that is the task.
-            pull : str or list of str 
-                The names of objects to be pulled as results.
-            push : dict
-                A dict of objects to be pushed into the engines namespace before
-                execution of the expression.
-            clear_before : boolean
-                Should the engine's namespace be cleared before the task is run.
-                Default=False.
-            clear_after : boolean 
-                Should the engine's namespace be cleared after the task is run.
-                Default=False.
-            retries : int
-                The number of times to resumbit the task if it fails.  Default=0.
-            options : dict
-                Any other keyword options for more elaborate uses of tasks
+            task : an `ITask` implementer
             
         :Returns: The int taskid of the submitted task.  Pass this to 
             `get_task_result` to get the `TaskResult` object.
         """
-        assert isinstance(task, taskmodule.Task), "task must be a Task object!"
-        ctask = taskmodule.canTask(task) # handles arbitrary function in .depend
-                                # as well as arbitrary recovery_task chains
-        ptask = pickle.dumps(ctask, 2)
+        assert isinstance(task, taskmodule.BaseTask), "task must be a Task object!"
+        task.can_task()
+        ptask = pickle.dumps(task, 2)
+        task.uncan_task()
         d = self.remote_reference.callRemote('run', ptask)
         d.addCallback(self.unpackage)
         return d
     
     def get_task_result(self, taskid, block=False):
-        """The task result by taskid.
+        """
+        Get a task result by taskid.
         
         :Parameters:
             taskid : int
@@ -224,20 +230,19 b' class FCTaskClient(object):'
         return d 
     
     def abort(self, taskid):
-        """Abort a task by taskid.
+        """
+        Abort a task by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be aborted.
-            block : boolean
-                Should I block until the task is aborted.        
         """
         d = self.remote_reference.callRemote('abort', taskid)
         d.addCallback(self.unpackage)
         return d 
         
     def barrier(self, taskids):
-        """Block until all tasks are completed.
+        """Block until a set of tasks are completed.
         
         :Parameters:
             taskids : list, tuple
@@ -248,20 +253,77 b' class FCTaskClient(object):'
         return d 
     
     def spin(self):
-        """touch the scheduler, to resume scheduling without submitting
-        a task.
+        """
+        Touch the scheduler, to resume scheduling without submitting a task.
+        
+        This method only needs to be called in unusual situations where the
+        scheduler is idle for some reason.
         """
         d = self.remote_reference.callRemote('spin')
         d.addCallback(self.unpackage)
         return d
     
     def queue_status(self, verbose=False):
-        """Return a dict with the status of the task queue."""
+        """
+        Get a dictionary with the current state of the task queue.
+        
+        :Parameters:
+            verbose : boolean
+                If True, return a list of taskids.  If False, simply give
+                the number of tasks with each status.
+        
+        :Returns:
+            A dict with the queue status.
+        """
         d = self.remote_reference.callRemote('queue_status', verbose)
         d.addCallback(self.unpackage)
         return d
     
+    def clear(self):
+        """
+        Clear all previously run tasks from the task controller.
+        
+        This is needed because the task controller keep all task results
+        in memory.  This can be a problem is there are many completed
+        tasks.  Users should call this periodically to clean out these
+        cached task results.
+        """
+        d = self.remote_reference.callRemote('clear')
+        return d
+    
     def adapt_to_blocking_client(self):
+        """
+        Wrap self in a blocking version that implements `IBlockingTaskClient.
+        """
         from IPython.kernel.taskclient import IBlockingTaskClient
         return IBlockingTaskClient(self)
 
+    def map(self, func, *sequences):
+        """
+        Apply func to *sequences elementwise.  Like Python's builtin map.
+        
+        This version is load balanced.
+        """
+        return self.mapper().map(func, *sequences)
+    
+    def mapper(self, clear_before=False, clear_after=False, retries=0, 
+                recovery_task=None, depend=None, block=True):
+        """
+        Create an `IMapper` implementer with a given set of arguments.
+        
+        The `IMapper` created using a task controller is load balanced.
+        
+        See the documentation for `IPython.kernel.task.BaseTask` for 
+        documentation on the arguments to this method.
+        """
+        return TaskMapper(self, clear_before=clear_before, 
+            clear_after=clear_after, retries=retries, 
+            recovery_task=recovery_task, depend=depend, block=block)
+    
+    def parallel(self, clear_before=False, clear_after=False, retries=0, 
+        recovery_task=None, depend=None, block=True):
+        mapper = self.mapper(clear_before, clear_after, retries,
+            recovery_task, depend, block)
+        pf = ParallelFunction(mapper)
+        return pf
+

IPython/kernel/tests/engineservicetest.py

@@ -163,7 +163,6 b' class IEngineCoreTestCase(object):'
         try:
             import numpy
         except:
-            print 'no numpy, ',
             return
         a = numpy.random.random(1000)
         d = self.engine.push(dict(a=a))

IPython/kernel/tests/multienginetest.py

@@ -750,16 +750,6 b' class ISynchronousMultiEngineCoordinatorTestCase(IMultiEngineCoordinatorTestCase'
             d.addCallback(lambda r: assert_array_equal(r, a))
             return d
 
-    def testMapNonblocking(self):
-        self.addEngine(4)
-        def f(x):
-            return x**2
-        data = range(16)
-        d= self.multiengine.map(f, data, block=False)
-        d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
-        d.addCallback(lambda r: self.assertEquals(r,[f(x) for x in data]))
-        return d
-
     def test_clear_pending_deferreds(self):
         self.addEngine(4)
         did_list = []

IPython/kernel/tests/tasktest.py

@@ -43,23 +43,23 b' class TaskTestBase(object):'
 
 class ITaskControllerTestCase(TaskTestBase):
         
-    def testTaskIDs(self):
+    def test_task_ids(self):
         self.addEngine(1)
-        d = self.tc.run(task.Task('a=5'))
+        d = self.tc.run(task.StringTask('a=5'))
         d.addCallback(lambda r: self.assertEquals(r, 0))
-        d.addCallback(lambda r: self.tc.run(task.Task('a=5')))
+        d.addCallback(lambda r: self.tc.run(task.StringTask('a=5')))
         d.addCallback(lambda r: self.assertEquals(r, 1))
-        d.addCallback(lambda r: self.tc.run(task.Task('a=5')))
+        d.addCallback(lambda r: self.tc.run(task.StringTask('a=5')))
         d.addCallback(lambda r: self.assertEquals(r, 2))
-        d.addCallback(lambda r: self.tc.run(task.Task('a=5')))
+        d.addCallback(lambda r: self.tc.run(task.StringTask('a=5')))
         d.addCallback(lambda r: self.assertEquals(r, 3))
         return d
     
-    def testAbort(self):
+    def test_abort(self):
         """Cannot do a proper abort test, because blocking execution prevents
         abort from being called before task completes"""
         self.addEngine(1)
-        t = task.Task('a=5')
+        t = task.StringTask('a=5')
         d = self.tc.abort(0)
         d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
         d.addCallback(lambda _:self.tc.run(t))
@@ -67,15 +67,15 b' class ITaskControllerTestCase(TaskTestBase):'
         d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
         return d
     
-    def testAbortType(self):
+    def test_abort_type(self):
         self.addEngine(1)
         d = self.tc.abort('asdfadsf')
         d.addErrback(lambda f: self.assertRaises(TypeError, f.raiseException))
         return d
     
-    def testClears(self):
+    def test_clear_before_and_after(self):
         self.addEngine(1)
-        t = task.Task('a=1', clear_before=True, pull='b', clear_after=True)
+        t = task.StringTask('a=1', clear_before=True, pull='b', clear_after=True)
         d = self.multiengine.execute('b=1', targets=0)
         d.addCallback(lambda _: self.tc.run(t))
         d.addCallback(lambda tid: self.tc.get_task_result(tid,block=True))
@@ -85,10 +85,10 b' class ITaskControllerTestCase(TaskTestBase):'
         d.addErrback(lambda f: self.assertRaises(NameError, _raise_it, f))
         return d
     
-    def testSimpleRetries(self):
+    def test_simple_retries(self):
         self.addEngine(1)
-        t = task.Task("i += 1\nassert i == 16", pull='i',retries=10)
-        t2 = task.Task("i += 1\nassert i == 16", pull='i',retries=10)
+        t = task.StringTask("i += 1\nassert i == 16", pull='i',retries=10)
+        t2 = task.StringTask("i += 1\nassert i == 16", pull='i',retries=10)
         d = self.multiengine.execute('i=0', targets=0)
         d.addCallback(lambda r: self.tc.run(t))
         d.addCallback(self.tc.get_task_result, block=True)
@@ -101,10 +101,10 b' class ITaskControllerTestCase(TaskTestBase):'
         d.addCallback(lambda r: self.assertEquals(r, 16))
         return d
     
-    def testRecoveryTasks(self):
+    def test_recovery_tasks(self):
         self.addEngine(1)
-        t = task.Task("i=16", pull='i')
-        t2 = task.Task("raise Exception", recovery_task=t, retries = 2)
+        t = task.StringTask("i=16", pull='i')
+        t2 = task.StringTask("raise Exception", recovery_task=t, retries = 2)
         
         d = self.tc.run(t2)
         d.addCallback(self.tc.get_task_result, block=True)
@@ -112,47 +112,76 b' class ITaskControllerTestCase(TaskTestBase):'
         d.addCallback(lambda r: self.assertEquals(r, 16))
         return d
     
-    # def testInfiniteRecoveryLoop(self):
-    #     self.addEngine(1)
-    #     t = task.Task("raise Exception", retries = 5)
-    #     t2 = task.Task("assert True", retries = 2, recovery_task = t)
-    #     t.recovery_task = t2
-    #     
-    #     d = self.tc.run(t)
-    #     d.addCallback(self.tc.get_task_result, block=True)
-    #     d.addCallback(lambda tr: tr.ns.i)
-    #     d.addBoth(printer)
-    #     d.addErrback(lambda f: self.assertRaises(AssertionError, f.raiseException))
-    #     return d
-    # 
-    def testSetupNS(self):
+    def test_setup_ns(self):
         self.addEngine(1)
         d = self.multiengine.execute('a=0', targets=0)
         ns = dict(a=1, b=0)
-        t = task.Task("", push=ns, pull=['a','b'])
+        t = task.StringTask("", push=ns, pull=['a','b'])
         d.addCallback(lambda r: self.tc.run(t))
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: {'a':tr.ns.a, 'b':tr['b']})
         d.addCallback(lambda r: self.assertEquals(r, ns))
         return d
     
-    def testTaskResults(self):
+    def test_string_task_results(self):
         self.addEngine(1)
-        t1 = task.Task('a=5', pull='a')
+        t1 = task.StringTask('a=5', pull='a')
         d = self.tc.run(t1)
         d.addCallback(self.tc.get_task_result, block=True)
-        d.addCallback(lambda tr: (tr.ns.a,tr['a'],tr.failure, tr.raiseException()))
+        d.addCallback(lambda tr: (tr.ns.a,tr['a'],tr.failure, tr.raise_exception()))
         d.addCallback(lambda r: self.assertEquals(r, (5,5,None,None)))  
         
-        t2 = task.Task('7=5')
+        t2 = task.StringTask('7=5')
         d.addCallback(lambda r: self.tc.run(t2))
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: tr.ns)
         d.addErrback(lambda f: self.assertRaises(SyntaxError, f.raiseException))
         
-        t3 = task.Task('', pull='b')
+        t3 = task.StringTask('', pull='b')
         d.addCallback(lambda r: self.tc.run(t3))
         d.addCallback(self.tc.get_task_result, block=True)
         d.addCallback(lambda tr: tr.ns)
         d.addErrback(lambda f: self.assertRaises(NameError, f.raiseException))
         return d
+    
+    def test_map_task(self):
+        self.addEngine(1)
+        t1 = task.MapTask(lambda x: 2*x,(10,))
+        d = self.tc.run(t1)
+        d.addCallback(self.tc.get_task_result, block=True)
+        d.addCallback(lambda r: self.assertEquals(r,20))
+        
+        t2 = task.MapTask(lambda : 20)
+        d.addCallback(lambda _: self.tc.run(t2))
+        d.addCallback(self.tc.get_task_result, block=True)
+        d.addCallback(lambda r: self.assertEquals(r,20))
+
+        t3 = task.MapTask(lambda x: x,(),{'x':20})
+        d.addCallback(lambda _: self.tc.run(t3))
+        d.addCallback(self.tc.get_task_result, block=True)
+        d.addCallback(lambda r: self.assertEquals(r,20))
+        return d
+    
+    def test_map_task_failure(self):
+        self.addEngine(1)
+        t1 = task.MapTask(lambda x: 1/0,(10,))
+        d = self.tc.run(t1)
+        d.addCallback(self.tc.get_task_result, block=True)
+        d.addErrback(lambda f: self.assertRaises(ZeroDivisionError, f.raiseException))
+        return d
+    
+    def test_map_task_args(self):
+        self.assertRaises(TypeError, task.MapTask, 'asdfasdf')
+        self.assertRaises(TypeError, task.MapTask, lambda x: x, 10)
+        self.assertRaises(TypeError, task.MapTask, lambda x: x, (10,),30)
+    
+    def test_clear(self):
+        self.addEngine(1)
+        t1 = task.MapTask(lambda x: 2*x,(10,))
+        d = self.tc.run(t1)
+        d.addCallback(lambda _: self.tc.get_task_result(0, block=True))
+        d.addCallback(lambda r: self.assertEquals(r,20))
+        d.addCallback(lambda _: self.tc.clear())
+        d.addCallback(lambda _: self.tc.get_task_result(0, block=True))
+        d.addErrback(lambda f: self.assertRaises(IndexError, f.raiseException))
+        return d

IPython/kernel/tests/test_enginefc.py

@@ -38,7 +38,7 b' try:'
         IEngineQueuedTestCase
 except ImportError:
     print "we got an error!!!"
-    pass
+    raise
 else:
     class EngineFCTest(DeferredTestCase, 
                        IEngineCoreTestCase, 

IPython/kernel/tests/test_multienginefc.py

@@ -26,9 +26,20 b' try:'
     from IPython.kernel.multienginefc import IFCSynchronousMultiEngine
     from IPython.kernel import multiengine as me
     from IPython.kernel.clientconnector import ClientConnector
+    from IPython.kernel.parallelfunction import ParallelFunction
+    from IPython.kernel.error import CompositeError
+    from IPython.kernel.util import printer
 except ImportError:
     pass
 else:
+    
+    def _raise_it(f):
+        try:
+            f.raiseException()
+        except CompositeError, e:
+            e.raise_exception()
+    
+    
     class FullSynchronousMultiEngineTestCase(DeferredTestCase, IFullSynchronousMultiEngineTestCase):
     
         def setUp(self):
@@ -68,3 +79,66 b' else:'
             d.addBoth(lambda _: self.controller.stopService())
             dlist.append(d)
             return defer.DeferredList(dlist)
+
+        def test_mapper(self):
+            self.addEngine(4)
+            m = self.multiengine.mapper()
+            self.assertEquals(m.multiengine,self.multiengine)
+            self.assertEquals(m.dist,'b')
+            self.assertEquals(m.targets,'all')
+            self.assertEquals(m.block,True)
+        
+        def test_map_default(self):
+            self.addEngine(4)
+            m = self.multiengine.mapper()
+            d = m.map(lambda x: 2*x, range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            d.addCallback(lambda _: self.multiengine.map(lambda x: 2*x, range(10)))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+        
+        def test_map_noblock(self):
+            self.addEngine(4)
+            m = self.multiengine.mapper(block=False)
+            d = m.map(lambda x: 2*x, range(10))
+            d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+                
+        def test_mapper_fail(self):
+            self.addEngine(4)
+            m = self.multiengine.mapper()
+            d = m.map(lambda x: 1/0, range(10))
+            d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
+            return d
+        
+        def test_parallel(self):
+            self.addEngine(4)
+            p = self.multiengine.parallel()
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 2*x
+            d = f(range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+        
+        def test_parallel_noblock(self):
+            self.addEngine(1)
+            p = self.multiengine.parallel(block=False)
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 2*x
+            d = f(range(10))
+            d.addCallback(lambda did: self.multiengine.get_pending_deferred(did, True))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+        
+        def test_parallel_fail(self):
+            self.addEngine(4)
+            p = self.multiengine.parallel()
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 1/0
+            d = f(range(10))
+            d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
+            return d
  No newline at end of file

IPython/kernel/tests/test_pendingdeferred.py

@@ -20,8 +20,6 b' try:'
     from twisted.internet import defer
     from twisted.python import failure
 
-    from IPython.testing import tcommon
-    from IPython.testing.tcommon import *
     from IPython.testing.util import DeferredTestCase
     import IPython.kernel.pendingdeferred as pd
     from IPython.kernel import error
@@ -30,25 +28,6 b' except ImportError:'
     pass
 else:
 
-    #-------------------------------------------------------------------------------
-    # Setup for inline and standalone doctests
-    #-------------------------------------------------------------------------------
-
-
-    # If you have standalone doctests in a separate file, set their names in the
-    # dt_files variable (as a single string  or a list thereof):
-    dt_files = []
-
-    # If you have any modules whose docstrings should be scanned for embedded tests
-    # as examples accorging to standard doctest practice, set them here (as a
-    # single string or a list thereof):
-    dt_modules = []
-
-    #-------------------------------------------------------------------------------
-    # Regular Unittests
-    #-------------------------------------------------------------------------------
-
-
     class Foo(object):
     
         def bar(self, bahz):
@@ -205,14 +184,3 b' else:'
             d3 = self.pdm.get_pending_deferred(did,False)
             d3.addCallback(lambda r: self.assertEquals(r,'bar'))
 
-#-------------------------------------------------------------------------------
-# Regular Unittests
-#-------------------------------------------------------------------------------
-
-# This ensures that the code will run either standalone as a script, or that it
-# can be picked up by Twisted's `trial` test wrapper to run all the tests.
-if tcommon.pexpect is not None:
-    if __name__ == '__main__':
-        unittest.main(testLoader=IPDocTestLoader(dt_files,dt_modules))
-    else:
-        testSuite = lambda : makeTestSuite(__name__,dt_files,dt_modules)

IPython/kernel/tests/test_taskfc.py

@@ -30,6 +30,8 b' try:'
     from IPython.kernel.util import printer
     from IPython.kernel.tests.tasktest import ITaskControllerTestCase
     from IPython.kernel.clientconnector import ClientConnector
+    from IPython.kernel.error import CompositeError
+    from IPython.kernel.parallelfunction import ParallelFunction
 except ImportError:
     pass
 else:
@@ -38,6 +40,12 b' else:'
     # Tests
     #-------------------------------------------------------------------------------
 
+    def _raise_it(f):
+        try:
+            f.raiseException()
+        except CompositeError, e:
+            e.raise_exception()
+
     class TaskTest(DeferredTestCase, ITaskControllerTestCase):
 
         def setUp(self):
@@ -88,3 +96,66 b' else:'
             dlist.append(d)
             return defer.DeferredList(dlist)
 
+        def test_mapper(self):
+            self.addEngine(1)
+            m = self.tc.mapper()
+            self.assertEquals(m.task_controller,self.tc)
+            self.assertEquals(m.clear_before,False)
+            self.assertEquals(m.clear_after,False)
+            self.assertEquals(m.retries,0)
+            self.assertEquals(m.recovery_task,None)
+            self.assertEquals(m.depend,None)
+            self.assertEquals(m.block,True)
+        
+        def test_map_default(self):
+            self.addEngine(1)
+            m = self.tc.mapper()
+            d = m.map(lambda x: 2*x, range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            d.addCallback(lambda _: self.tc.map(lambda x: 2*x, range(10)))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+        
+        def test_map_noblock(self):
+            self.addEngine(1)
+            m = self.tc.mapper(block=False)
+            d = m.map(lambda x: 2*x, range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[x for x in range(10)]))
+            return d
+                
+        def test_mapper_fail(self):
+            self.addEngine(1)
+            m = self.tc.mapper()
+            d = m.map(lambda x: 1/0, range(10))
+            d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
+            return d
+        
+        def test_parallel(self):
+            self.addEngine(1)
+            p = self.tc.parallel()
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 2*x
+            d = f(range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[2*x for x in range(10)]))
+            return d
+        
+        def test_parallel_noblock(self):
+            self.addEngine(1)
+            p = self.tc.parallel(block=False)
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 2*x
+            d = f(range(10))
+            d.addCallback(lambda r: self.assertEquals(r,[x for x in range(10)]))
+            return d
+        
+        def test_parallel_fail(self):
+            self.addEngine(1)
+            p = self.tc.parallel()
+            self.assert_(isinstance(p, ParallelFunction))
+            @p
+            def f(x): return 1/0
+            d = f(range(10))
+            d.addBoth(lambda f: self.assertRaises(ZeroDivisionError, _raise_it, f))
+            return d
  No newline at end of file

IPython/platutils.py

@@ -3,13 +3,8 b''
 
 Importing this module should give you the implementations that are correct 
 for your operation system, from platutils_PLATFORMNAME module.
-
-$Id: ipstruct.py 1005 2006-01-12 08:39:26Z fperez $
-
-
 """
 
-
 #*****************************************************************************
 #       Copyright (C) 2001-2006 Fernando Perez <[e-mail]>
 #
@@ -21,15 +16,32 b' from IPython import Release'
 __author__  = '%s <%s>' % Release.authors['Ville']
 __license__ = Release.license
 
-import os,sys
+import os
+import sys
 
+# Import the platform-specific implementations
 if os.name == 'posix':
-    from platutils_posix import *
+    import platutils_posix as _platutils
 elif sys.platform == 'win32':
-    from platutils_win32 import *
+    import platutils_win32 as _platutils
 else:
-    from platutils_dummy import *
+    import platutils_dummy as _platutils
     import warnings
     warnings.warn("Platutils not available for platform '%s', some features may be missing" %
         os.name)
     del warnings
+
+
+# Functionality that's logically common to all platforms goes here, each
+# platform-specific module only provides the bits that are OS-dependent.
+
+def freeze_term_title():
+    _platutils.ignore_termtitle = True
+
+
+def set_term_title(title):
+    """Set terminal title using the necessary platform-dependent calls."""
+
+    if _platutils.ignore_termtitle:
+        return
+    _platutils.set_term_title(title)

IPython/platutils_dummy.py

@@ -3,13 +3,8 b''
 
 This has empty implementation of the platutils functions, used for 
 unsupported operating systems.
-
-$Id: ipstruct.py 1005 2006-01-12 08:39:26Z fperez $
-
-
 """
 
-
 #*****************************************************************************
 #       Copyright (C) 2001-2006 Fernando Perez <[e-mail]>
 #
@@ -21,9 +16,9 b' from IPython import Release'
 __author__  = '%s <%s>' % Release.authors['Ville']
 __license__ = Release.license
 
+# This variable is part of the expected API of the module:
+ignore_termtitle = True
 
-def _dummy(*args,**kw):
+def set_term_title(*args,**kw):
+    """Dummy no-op."""
     pass
-
-set_term_title = _dummy
-

IPython/platutils_posix.py

@@ -3,12 +3,8 b''
 
 Importing this module directly is not portable - rather, import platutils 
 to use these functions in platform agnostic fashion.
-
-$Id: ipstruct.py 1005 2006-01-12 08:39:26Z fperez $
-
 """
 
-
 #*****************************************************************************
 #       Copyright (C) 2001-2006 Fernando Perez <[e-mail]>
 #
@@ -31,9 +27,6 b' def _dummy_op(*a, **b):'
 def _set_term_title_xterm(title):
     """ Change virtual terminal title in xterm-workalikes """
 
-    if ignore_termtitle:
-        return
-    
     sys.stdout.write('\033]%d;%s\007' % (0,title))
 
 
@@ -41,7 +34,3 b" if os.environ.get('TERM','') == 'xterm':"
     set_term_title = _set_term_title_xterm
 else:
     set_term_title = _dummy_op
-
-def freeze_term_title():
-    global ignore_termtitle
-    ignore_termtitle = True

IPython/platutils_win32.py

@@ -3,12 +3,8 b''
 
 Importing this module directly is not portable - rather, import platutils 
 to use these functions in platform agnostic fashion.
-
-$Id: ipstruct.py 1005 2006-01-12 08:39:26Z fperez $
-
 """
 
-
 #*****************************************************************************
 #       Copyright (C) 2001-2006 Fernando Perez <[e-mail]>
 #
@@ -22,35 +18,30 b' __license__ = Release.license'
 
 import os
 
-ignore_termtitle = 0
+ignore_termtitle = False
 
 try:
     import ctypes
+
     SetConsoleTitleW=ctypes.windll.kernel32.SetConsoleTitleW
     SetConsoleTitleW.argtypes=[ctypes.c_wchar_p]
-    def _set_term_title(title):
-        """ Set terminal title using the ctypes"""
+    
+    def set_term_title(title):
+        """Set terminal title using ctypes to access the Win32 APIs."""
         SetConsoleTitleW(title)
 
 except ImportError:
-    def _set_term_title(title):
-        """ Set terminal title using the 'title' command """
+    def set_term_title(title):
+        """Set terminal title using the 'title' command."""
+        global ignore_termtitle
+
+        try:
+            # Cannot be on network share when issuing system commands
         curr=os.getcwd()
-        os.chdir("C:") #Cannot be on network share when issuing system commands
+            os.chdir("C:")
         ret = os.system("title " + title)
+        finally:
         os.chdir(curr)
         if ret:
-            ignore_termtitle = 1
-
-def set_term_title(title):
-    """ Set terminal title using the 'title' command """
-    global ignore_termtitle
-
-    if ignore_termtitle:
-        return
-    _set_term_title(title)
-
-def freeze_term_title():
-    global ignore_termtitle
-    ignore_termtitle = 1
-
+            # non-zero return code signals error, don't try again
+            ignore_termtitle = True

IPython/testing/tutils.py

@@ -1,6 +1,20 b''
-"""Utilities for testing code.
+"""DEPRECATED - use IPython.testing.util instead.
+
+Utilities for testing code.
 """
 
+#############################################################################
+
+# This was old testing code we never really used in IPython.  The pieces of
+# testing machinery from snakeoil that were good have already been merged into
+# the nose plugin, so this can be taken away soon.  Leave a warning for now,
+# we'll remove it in a later release (around 0.10 or so).
+from warnings import warn
+warn('This will be removed soon.  Use IPython.testing.util instead',
+     DeprecationWarning)
+
+#############################################################################
+
 # Required modules and packages
 
 # Standard Python lib

IPython/tools/tests/__init__.py

@@ -1,10 +0,0 b''
-# encoding: utf-8
-__docformat__ = "restructuredtext en"
-#-------------------------------------------------------------------------------
-#       Copyright (C) 2005  Fernando Perez <[e-mail]>
-#                           Brian E Granger <[e-mail]>
-#                           Benjamin Ragan-Kelley <[e-mail]>
-#
-#  Distributed under the terms of the BSD License.  The full license is in
-#  the file COPYING, distributed as part of this software.
-#-------------------------------------------------------------------------------

IPython/tools/tests/test_tools_utils.txt

@@ -3,22 +3,8 b''
 =========================================
 
 The way doctest loads these, the entire document is applied as a single test
-rather than multiple individual ones, unfortunately.
+rather than multiple individual ones, unfortunately::
 
-
-Auto-generated tests
-====================
-
-
-----------------------------------------------------------------------------
-
-Begin included file tst_tools_utils_doctest2.py::
-
-    # Setup - all imports are done in tcommon
-    >>> from IPython.testing import tcommon
-    >>> from IPython.testing.tcommon import *
-    
-    # Doctest code begins here
     >>> from IPython.tools import utils
     
     # Some other tests for utils
@@ -29,14 +15,3 b' Begin included file tst_tools_utils_doctest2.py::'
     >>> utils.marquee('Another test',30,'.')
     '........ Another test ........'
     
-    
-End included file tst_tools_utils_doctest2.py
-
-----------------------------------------------------------------------------
-
-
-
-Manually generated tests
-========================
-
-These are one-off tests written by hand, copied from an interactive prompt.

docs/emacs/ipython.el

@@ -31,13 +31,20 b''
 ;; To start an interactive ipython session run `py-shell' with ``M-x py-shell``
 ;; (or the default keybinding ``C-c C-!``).
 ;;
+;; You can customize the arguments passed to the IPython instance at startup by
+;; setting the ``py-python-command-args`` variable.  For example, to start
+;; always in ``pylab`` mode with hardcoded light-background colors, you can
+;; use::
+;;
+;; (setq py-python-command-args '("-pylab" "-colors" "LightBG"))
+;;
+;;
 ;; NOTE: This mode is currently somewhat alpha and although I hope that it
 ;; will work fine for most cases, doing certain things (like the
 ;; autocompletion and a decent scheme to switch between python interpreters)
 ;; properly will also require changes to ipython that will likely have to wait
 ;; for a larger rewrite scheduled some time in the future.
 ;; 
-;; Also note that you currently NEED THE CVS VERSION OF PYTHON.EL.
 ;;
 ;; Further note that I don't know whether this runs under windows or not and
 ;; that if it doesn't I can't really help much, not being afflicted myself.
@@ -46,14 +53,15 b''
 ;; Hints for effective usage
 ;; -------------------------
 ;;
-;; - IMO the best feature by far of the ipython/emacs combo is how much easier it
-;;   makes it to find and fix bugs thanks to the ``%pdb on``/ pdbtrack combo. Try
-;;   it: first in the ipython to shell do ``%pdb on`` then do something that will
-;;   raise an exception (FIXME nice example) -- and be amazed how easy it is to
-;;   inspect the live objects in each stack frames and to jump to the
-;;   corresponding sourcecode locations as you walk up and down the stack trace
-;;   (even without ``%pdb on`` you can always use ``C-c -`` (`py-up-exception')
-;;   to jump to the corresponding source code locations).
+;; - IMO the best feature by far of the ipython/emacs combo is how much easier
+;;   it makes it to find and fix bugs thanks to the ``%pdb on or %debug``/
+;;   pdbtrack combo. Try it: first in the ipython to shell do ``%pdb on`` then
+;;   do something that will raise an exception (FIXME nice example), or type
+;;   ``%debug`` after the exception has been raised.  YOu'll be amazed at how
+;;   easy it is to inspect the live objects in each stack frames and to jump to
+;;   the corresponding sourcecode locations as you walk up and down the stack
+;;   trace (even without ``%pdb on`` you can always use ``C-c -``
+;;   (`py-up-exception') to jump to the corresponding source code locations).
 ;;
 ;; - emacs gives you much more powerful commandline editing and output searching
 ;;   capabilities than ipython-standalone -- isearch is your friend if you
@@ -79,7 +87,7 b''
 ;;   variables comes later).
 ;;
 ;; Please send comments and feedback to the ipython-list
-;; (<[e-mail]>) where I (a.s.) or someone else will try to
+;; (<[e-mail]>) where I (a.s.) or someone else will try to
 ;; answer them (it helps if you specify your emacs version, OS etc; 
 ;; familiarity with <http://www.catb.org/~esr/faqs/smart-questions.html> might
 ;; speed up things further).