Merge pull request #2871 from minrk/shortexc

truncate potentially long CompositeErrors

IPython/parallel/error.py

@@ -1,346 +1,354 b''
 # encoding: utf-8
 
 """Classes and functions for kernel related errors and exceptions.
 
 Inheritance diagram:
 
 .. inheritance-diagram:: IPython.parallel.error
    :parts: 3
 
 Authors:
 
 * Brian Granger
 * Min RK
 """
 from __future__ import print_function
 
 import sys
 import traceback
 
 __docformat__ = "restructuredtext en"
 
 # Tell nose to skip this module
 __test__ = {}
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  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.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Error classes
 #-------------------------------------------------------------------------------
 class IPythonError(Exception):
     """Base exception that all of our exceptions inherit from.
 
     This can be raised by code that doesn't have any more specific
     information."""
 
     pass
 
 # Exceptions associated with the controller objects
 class ControllerError(IPythonError): pass
 
 class ControllerCreationError(ControllerError): pass
 
 
 # Exceptions associated with the Engines
 class EngineError(IPythonError): pass
 
 class EngineCreationError(EngineError): pass
 
 class KernelError(IPythonError):
     pass
 
 class NotDefined(KernelError):
     def __init__(self, name):
         self.name = name
         self.args = (name,)
 
     def __repr__(self):
         return '<NotDefined: %s>' % self.name
 
     __str__ = __repr__
 
 
 class QueueCleared(KernelError):
     pass
 
 
 class IdInUse(KernelError):
     pass
 
 
 class ProtocolError(KernelError):
     pass
 
 
 class ConnectionError(KernelError):
     pass
 
 
 class InvalidEngineID(KernelError):
     pass
 
 
 class NoEnginesRegistered(KernelError):
     pass
 
 
 class InvalidClientID(KernelError):
     pass
 
 
 class InvalidDeferredID(KernelError):
     pass
 
 
 class SerializationError(KernelError):
     pass
 
 
 class MessageSizeError(KernelError):
     pass
 
 
 class PBMessageSizeError(MessageSizeError):
     pass
 
 
 class ResultNotCompleted(KernelError):
     pass
 
 
 class ResultAlreadyRetrieved(KernelError):
     pass
 
 class ClientError(KernelError):
     pass
 
 
 class TaskAborted(KernelError):
     pass
 
 
 class TaskTimeout(KernelError):
     pass
 
 
 class NotAPendingResult(KernelError):
     pass
 
 
 class UnpickleableException(KernelError):
     pass
 
 
 class AbortedPendingDeferredError(KernelError):
     pass
 
 
 class InvalidProperty(KernelError):
     pass
 
 
 class MissingBlockArgument(KernelError):
     pass
 
 
 class StopLocalExecution(KernelError):
     pass
 
 
 class SecurityError(KernelError):
     pass
 
 
 class FileTimeoutError(KernelError):
     pass
 
 class TimeoutError(KernelError):
     pass
 
 class UnmetDependency(KernelError):
     pass
 
 class ImpossibleDependency(UnmetDependency):
     pass
 
 class DependencyTimeout(ImpossibleDependency):
     pass
 
 class InvalidDependency(ImpossibleDependency):
     pass
 
 class RemoteError(KernelError):
     """Error raised elsewhere"""
     ename=None
     evalue=None
     traceback=None
     engine_info=None
 
     def __init__(self, ename, evalue, traceback, engine_info=None):
         self.ename=ename
         self.evalue=evalue
         self.traceback=traceback
         self.engine_info=engine_info or {}
         self.args=(ename, evalue)
 
     def __repr__(self):
         engineid = self.engine_info.get('engine_id', ' ')
         return "<Remote[%s]:%s(%s)>"%(engineid, self.ename, self.evalue)
 
     def __str__(self):
         return "%s(%s)" % (self.ename, self.evalue)
     
     def render_traceback(self):
         """render traceback to a list of lines"""
         return (self.traceback or "No traceback available").splitlines()
     
     def _render_traceback_(self):
         """Special method for custom tracebacks within IPython.
         
         This will be called by IPython instead of displaying the local traceback.
         
         It should return a traceback rendered as a list of lines.
         """
         return self.render_traceback()
     
     def print_traceback(self, excid=None):
         """print my traceback"""
         print('\n'.join(self.render_traceback()))
 
     
 
 
 class TaskRejectError(KernelError):
     """Exception to raise when a task should be rejected by an engine.
 
     This exception can be used to allow a task running on an engine to test
     if the engine (or the user's namespace on the engine) has the needed
     task dependencies.  If not, the task should raise this exception.  For
     the task to be retried on another engine, the task should be created
     with the `retries` argument > 1.
 
     The advantage of this approach over our older properties system is that
     tasks have full access to the user's namespace on the engines and the
     properties don't have to be managed or tested by the controller.
     """
 
 
 class CompositeError(RemoteError):
     """Error for representing possibly multiple errors on engines"""
+    tb_limit = 4 # limit on how many tracebacks to draw
+    
     def __init__(self, message, elist):
         Exception.__init__(self, *(message, elist))
         # Don't use pack_exception because it will conflict with the .message
         # attribute that is being deprecated in 2.6 and beyond.
         self.msg = message
         self.elist = elist
         self.args = [ e[0] for e in elist ]
 
     def _get_engine_str(self, ei):
         if not ei:
             return '[Engine Exception]'
         else:
             return '[%s:%s]: ' % (ei['engine_id'], ei['method'])
 
     def _get_traceback(self, ev):
         try:
             tb = ev._ipython_traceback_text
         except AttributeError:
             return 'No traceback available'
         else:
             return tb
 
     def __str__(self):
         s = str(self.msg)
-        for en, ev, etb, ei in self.elist:
+        for en, ev, etb, ei in self.elist[:self.tb_limit]:
             engine_str = self._get_engine_str(ei)
             s = s + '\n' + engine_str + en + ': ' + str(ev)
+        if len(self.elist) > self.tb_limit:
+            s = s + '\n.... %i more exceptions ...' % (len(self.elist) - self.tb_limit)
         return s
 
     def __repr__(self):
-        return "CompositeError(%i)"%len(self.elist)
+        return "CompositeError(%i)" % len(self.elist)
     
     def render_traceback(self, excid=None):
         """render one or all of my tracebacks to a list of lines"""
         lines = []
         if excid is None:
-            for (en,ev,etb,ei) in self.elist:
+            for (en,ev,etb,ei) in self.elist[:self.tb_limit]:
                 lines.append(self._get_engine_str(ei))
                 lines.extend((etb or 'No traceback available').splitlines())
                 lines.append('')
+            if len(self.elist) > self.tb_limit:
+                lines.append(
+                    '... %i more exceptions ...' % (len(self.elist) - self.tb_limit)
+                )
         else:
             try:
                 en,ev,etb,ei = self.elist[excid]
             except:
                 raise IndexError("an exception with index %i does not exist"%excid)
             else:
                 lines.append(self._get_engine_str(ei))
                 lines.extend((etb or 'No traceback available').splitlines())
         
         return lines
     
     def print_traceback(self, excid=None):
         print('\n'.join(self.render_traceback(excid)))
 
     def raise_exception(self, excid=0):
         try:
             en,ev,etb,ei = self.elist[excid]
         except:
             raise IndexError("an exception with index %i does not exist"%excid)
         else:
             raise RemoteError(en, ev, etb, ei)
 
 
 def collect_exceptions(rdict_or_list, method='unspecified'):
     """check a result dict for errors, and raise CompositeError if any exist.
     Passthrough otherwise."""
     elist = []
     if isinstance(rdict_or_list, dict):
         rlist = rdict_or_list.values()
     else:
         rlist = rdict_or_list
     for r in rlist:
         if isinstance(r, RemoteError):
             en, ev, etb, ei = r.ename, r.evalue, r.traceback, r.engine_info
             # Sometimes we could have CompositeError in our list.  Just take
             # the errors out of them and put them in our new list.  This
             # has the effect of flattening lists of CompositeErrors into one
             # CompositeError
             if en=='CompositeError':
                 for e in ev.elist:
                     elist.append(e)
             else:
                 elist.append((en, ev, etb, ei))
     if len(elist)==0:
         return rdict_or_list
     else:
         msg = "one or more exceptions from call to method: %s" % (method)
         # This silliness is needed so the debugger has access to the exception
         # instance (e in this case)
         try:
             raise CompositeError(msg, elist)
         except CompositeError as e:
             raise e
 
 def wrap_exception(engine_info={}):
     etype, evalue, tb = sys.exc_info()
     stb = traceback.format_exception(etype, evalue, tb)
     exc_content = {
         'status' : 'error',
         'traceback' : stb,
         'ename' : unicode(etype.__name__),
         'evalue' : unicode(evalue),
         'engine_info' : engine_info
     }
     return exc_content
 
 def unwrap_exception(content):
     err = RemoteError(content['ename'], content['evalue'],
                 ''.join(content['traceback']),
                 content.get('engine_info', {}))
     return err
 

IPython/parallel/tests/test_view.py

@@ -1,759 +1,786 b''
 # -*- coding: utf-8 -*-
 """test View objects
 
 Authors:
 
 * Min RK
 """
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import sys
 import platform
 import time
 from collections import namedtuple
 from tempfile import mktemp
 from StringIO import StringIO
 
 import zmq
 from nose import SkipTest
 from nose.plugins.attrib import attr
 
 from IPython.testing import decorators as dec
 from IPython.testing.ipunittest import ParametricTestCase
 from IPython.utils.io import capture_output
 
 from IPython import parallel  as pmod
 from IPython.parallel import error
 from IPython.parallel import AsyncResult, AsyncHubResult, AsyncMapResult
 from IPython.parallel import DirectView
 from IPython.parallel.util import interactive
 
 from IPython.parallel.tests import add_engines
 
 from .clienttest import ClusterTestCase, crash, wait, skip_without
 
 def setup():
     add_engines(3, total=True)
 
 point = namedtuple("point", "x y")
 
 class TestView(ClusterTestCase, ParametricTestCase):
     
     def setUp(self):
         # On Win XP, wait for resource cleanup, else parallel test group fails
         if platform.system() == "Windows" and platform.win32_ver()[0] == "XP":
             # 1 sec fails. 1.5 sec seems ok. Using 2 sec for margin of safety
             time.sleep(2)
         super(TestView, self).setUp()
 
     @attr('crash')
     def test_z_crash_mux(self):
         """test graceful handling of engine death (direct)"""
         # self.add_engines(1)
         eid = self.client.ids[-1]
         ar = self.client[eid].apply_async(crash)
         self.assertRaisesRemote(error.EngineError, ar.get, 10)
         eid = ar.engine_id
         tic = time.time()
         while eid in self.client.ids and time.time()-tic < 5:
             time.sleep(.01)
             self.client.spin()
         self.assertFalse(eid in self.client.ids, "Engine should have died")
     
     def test_push_pull(self):
         """test pushing and pulling"""
         data = dict(a=10, b=1.05, c=range(10), d={'e':(1,2),'f':'hi'})
         t = self.client.ids[-1]
         v = self.client[t]
         push = v.push
         pull = v.pull
         v.block=True
         nengines = len(self.client)
         push({'data':data})
         d = pull('data')
         self.assertEqual(d, data)
         self.client[:].push({'data':data})
         d = self.client[:].pull('data', block=True)
         self.assertEqual(d, nengines*[data])
         ar = push({'data':data}, block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         r = ar.get()
         ar = self.client[:].pull('data', block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         r = ar.get()
         self.assertEqual(r, nengines*[data])
         self.client[:].push(dict(a=10,b=20))
         r = self.client[:].pull(('a','b'), block=True)
         self.assertEqual(r, nengines*[[10,20]])
     
     def test_push_pull_function(self):
         "test pushing and pulling functions"
         def testf(x):
             return 2.0*x
         
         t = self.client.ids[-1]
         v = self.client[t]
         v.block=True
         push = v.push
         pull = v.pull
         execute = v.execute
         push({'testf':testf})
         r = pull('testf')
         self.assertEqual(r(1.0), testf(1.0))
         execute('r = testf(10)')
         r = pull('r')
         self.assertEqual(r, testf(10))
         ar = self.client[:].push({'testf':testf}, block=False)
         ar.get()
         ar = self.client[:].pull('testf', block=False)
         rlist = ar.get()
         for r in rlist:
             self.assertEqual(r(1.0), testf(1.0))
         execute("def g(x): return x*x")
         r = pull(('testf','g'))
         self.assertEqual((r[0](10),r[1](10)), (testf(10), 100))
     
     def test_push_function_globals(self):
         """test that pushed functions have access to globals"""
         @interactive
         def geta():
             return a
         # self.add_engines(1)
         v = self.client[-1]
         v.block=True
         v['f'] = geta
         self.assertRaisesRemote(NameError, v.execute, 'b=f()')
         v.execute('a=5')
         v.execute('b=f()')
         self.assertEqual(v['b'], 5)
     
     def test_push_function_defaults(self):
         """test that pushed functions preserve default args"""
         def echo(a=10):
             return a
         v = self.client[-1]
         v.block=True
         v['f'] = echo
         v.execute('b=f()')
         self.assertEqual(v['b'], 10)
 
     def test_get_result(self):
         """test getting results from the Hub."""
         c = pmod.Client(profile='iptest')
         # self.add_engines(1)
         t = c.ids[-1]
         v = c[t]
         v2 = self.client[t]
         ar = v.apply_async(wait, 1)
         # give the monitor time to notice the message
         time.sleep(.25)
         ahr = v2.get_result(ar.msg_ids)
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEqual(ahr.get(), ar.get())
         ar2 = v2.get_result(ar.msg_ids)
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.spin()
         c.close()
     
     def test_run_newline(self):
         """test that run appends newline to files"""
         tmpfile = mktemp()
         with open(tmpfile, 'w') as f:
             f.write("""def g():
                 return 5
                 """)
         v = self.client[-1]
         v.run(tmpfile, block=True)
         self.assertEqual(v.apply_sync(lambda f: f(), pmod.Reference('g')), 5)
 
     def test_apply_tracked(self):
         """test tracking for apply"""
         # self.add_engines(1)
         t = self.client.ids[-1]
         v = self.client[t]
         v.block=False
         def echo(n=1024*1024, **kwargs):
             with v.temp_flags(**kwargs):
                 return v.apply(lambda x: x, 'x'*n)
         ar = echo(1, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         ar = echo(track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertEqual(ar.sent, ar._tracker.done)
         ar._tracker.wait()
         self.assertTrue(ar.sent)
         
     def test_push_tracked(self):
         t = self.client.ids[-1]
         ns = dict(x='x'*1024*1024)
         v = self.client[t]
         ar = v.push(ns, block=False, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         
         ar = v.push(ns, block=False, track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         ar._tracker.wait()
         self.assertEqual(ar.sent, ar._tracker.done)
         self.assertTrue(ar.sent)
         ar.get()
         
     def test_scatter_tracked(self):
         t = self.client.ids
         x='x'*1024*1024
         ar = self.client[t].scatter('x', x, block=False, track=False)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertTrue(ar.sent)
         
         ar = self.client[t].scatter('x', x, block=False, track=True)
         self.assertTrue(isinstance(ar._tracker, zmq.MessageTracker))
         self.assertEqual(ar.sent, ar._tracker.done)
         ar._tracker.wait()
         self.assertTrue(ar.sent)
         ar.get()
     
     def test_remote_reference(self):
         v = self.client[-1]
         v['a'] = 123
         ra = pmod.Reference('a')
         b = v.apply_sync(lambda x: x, ra)
         self.assertEqual(b, 123)
 
 
     def test_scatter_gather(self):
         view = self.client[:]
         seq1 = range(16)
         view.scatter('a', seq1)
         seq2 = view.gather('a', block=True)
         self.assertEqual(seq2, seq1)
         self.assertRaisesRemote(NameError, view.gather, 'asdf', block=True)
     
     @skip_without('numpy')
     def test_scatter_gather_numpy(self):
         import numpy
         from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         view = self.client[:]
         a = numpy.arange(64)
         view.scatter('a', a, block=True)
         b = view.gather('a', block=True)
         assert_array_equal(b, a)
     
     def test_scatter_gather_lazy(self):
         """scatter/gather with targets='all'"""
         view = self.client.direct_view(targets='all')
         x = range(64)
         view.scatter('x', x)
         gathered = view.gather('x', block=True)
         self.assertEqual(gathered, x)
         
 
     @dec.known_failure_py3
     @skip_without('numpy')
     def test_push_numpy_nocopy(self):
         import numpy
         view = self.client[:]
         a = numpy.arange(64)
         view['A'] = a
         @interactive
         def check_writeable(x):
             return x.flags.writeable
         
         for flag in view.apply_sync(check_writeable, pmod.Reference('A')):
             self.assertFalse(flag, "array is writeable, push shouldn't have pickled it")
         
         view.push(dict(B=a))
         for flag in view.apply_sync(check_writeable, pmod.Reference('B')):
             self.assertFalse(flag, "array is writeable, push shouldn't have pickled it")
     
     @skip_without('numpy')
     def test_apply_numpy(self):
         """view.apply(f, ndarray)"""
         import numpy
         from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         
         A = numpy.random.random((100,100))
         view = self.client[-1]
         for dt in [ 'int32', 'uint8', 'float32', 'float64' ]:
             B = A.astype(dt)
             C = view.apply_sync(lambda x:x, B)
             assert_array_equal(B,C)
     
     @skip_without('numpy')
     def test_push_pull_recarray(self):
         """push/pull recarrays"""
         import numpy
         from numpy.testing.utils import assert_array_equal
         
         view = self.client[-1]
         
         R = numpy.array([
             (1, 'hi', 0.),
             (2**30, 'there', 2.5),
             (-99999, 'world', -12345.6789),
         ], [('n', int), ('s', '|S10'), ('f', float)])
         
         view['RR'] = R
         R2 = view['RR']
         
         r_dtype, r_shape = view.apply_sync(interactive(lambda : (RR.dtype, RR.shape)))
         self.assertEqual(r_dtype, R.dtype)
         self.assertEqual(r_shape, R.shape)
         self.assertEqual(R2.dtype, R.dtype)
         self.assertEqual(R2.shape, R.shape)
         assert_array_equal(R2, R)
 
     @skip_without('pandas')
     def test_push_pull_timeseries(self):
         """push/pull pandas.TimeSeries"""
         import pandas
         
         ts = pandas.TimeSeries(range(10))
         
         view = self.client[-1]
         
         view.push(dict(ts=ts), block=True)
         rts = view['ts']
         
         self.assertEqual(type(rts), type(ts))
         self.assertTrue((ts == rts).all())
     
     def test_map(self):
         view = self.client[:]
         def f(x):
             return x**2
         data = range(16)
         r = view.map_sync(f, data)
         self.assertEqual(r, map(f, data))
     
     def test_map_iterable(self):
         """test map on iterables (direct)"""
         view = self.client[:]
         # 101 is prime, so it won't be evenly distributed
         arr = range(101)
         # ensure it will be an iterator, even in Python 3
         it = iter(arr)
         r = view.map_sync(lambda x:x, arr)
         self.assertEqual(r, list(arr))
     
     def test_scatter_gather_nonblocking(self):
         data = range(16)
         view = self.client[:]
         view.scatter('a', data, block=False)
         ar = view.gather('a', block=False)
         self.assertEqual(ar.get(), data)
     
     @skip_without('numpy')
     def test_scatter_gather_numpy_nonblocking(self):
         import numpy
         from numpy.testing.utils import assert_array_equal, assert_array_almost_equal
         a = numpy.arange(64)
         view = self.client[:]
         ar = view.scatter('a', a, block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         amr = view.gather('a', block=False)
         self.assertTrue(isinstance(amr, AsyncMapResult))
         assert_array_equal(amr.get(), a)
 
     def test_execute(self):
         view = self.client[:]
         # self.client.debug=True
         execute = view.execute
         ar = execute('c=30', block=False)
         self.assertTrue(isinstance(ar, AsyncResult))
         ar = execute('d=[0,1,2]', block=False)
         self.client.wait(ar, 1)
         self.assertEqual(len(ar.get()), len(self.client))
         for c in view['c']:
             self.assertEqual(c, 30)
     
     def test_abort(self):
         view = self.client[-1]
         ar = view.execute('import time; time.sleep(1)', block=False)
         ar2 = view.apply_async(lambda : 2)
         ar3 = view.apply_async(lambda : 3)
         view.abort(ar2)
         view.abort(ar3.msg_ids)
         self.assertRaises(error.TaskAborted, ar2.get)
         self.assertRaises(error.TaskAborted, ar3.get)
     
     def test_abort_all(self):
         """view.abort() aborts all outstanding tasks"""
         view = self.client[-1]
         ars = [ view.apply_async(time.sleep, 0.25) for i in range(10) ]
         view.abort()
         view.wait(timeout=5)
         for ar in ars[5:]:
             self.assertRaises(error.TaskAborted, ar.get)
     
     def test_temp_flags(self):
         view = self.client[-1]
         view.block=True
         with view.temp_flags(block=False):
             self.assertFalse(view.block)
         self.assertTrue(view.block)
     
     @dec.known_failure_py3
     def test_importer(self):
         view = self.client[-1]
         view.clear(block=True)
         with view.importer:
             import re
         
         @interactive
         def findall(pat, s):
             # this globals() step isn't necessary in real code
             # only to prevent a closure in the test
             re = globals()['re']
             return re.findall(pat, s)
         
         self.assertEqual(view.apply_sync(findall, '\w+', 'hello world'), 'hello world'.split())
     
     def test_unicode_execute(self):
         """test executing unicode strings"""
         v = self.client[-1]
         v.block=True
         if sys.version_info[0] >= 3:
             code="a='é'"
         else:
             code=u"a=u'é'"
         v.execute(code)
         self.assertEqual(v['a'], u'é')
         
     def test_unicode_apply_result(self):
         """test unicode apply results"""
         v = self.client[-1]
         r = v.apply_sync(lambda : u'é')
         self.assertEqual(r, u'é')
     
     def test_unicode_apply_arg(self):
         """test passing unicode arguments to apply"""
         v = self.client[-1]
         
         @interactive
         def check_unicode(a, check):
             assert isinstance(a, unicode), "%r is not unicode"%a
             assert isinstance(check, bytes), "%r is not bytes"%check
             assert a.encode('utf8') == check, "%s != %s"%(a,check)
         
         for s in [ u'é', u'ßø®∫',u'asdf' ]:
             try:
                 v.apply_sync(check_unicode, s, s.encode('utf8'))
             except error.RemoteError as e:
                 if e.ename == 'AssertionError':
                     self.fail(e.evalue)
                 else:
                     raise e
     
     def test_map_reference(self):
         """view.map(<Reference>, *seqs) should work"""
         v = self.client[:]
         v.scatter('n', self.client.ids, flatten=True)
         v.execute("f = lambda x,y: x*y")
         rf = pmod.Reference('f')
         nlist = list(range(10))
         mlist = nlist[::-1]
         expected = [ m*n for m,n in zip(mlist, nlist) ]
         result = v.map_sync(rf, mlist, nlist)
         self.assertEqual(result, expected)
 
     def test_apply_reference(self):
         """view.apply(<Reference>, *args) should work"""
         v = self.client[:]
         v.scatter('n', self.client.ids, flatten=True)
         v.execute("f = lambda x: n*x")
         rf = pmod.Reference('f')
         result = v.apply_sync(rf, 5)
         expected = [ 5*id for id in self.client.ids ]
         self.assertEqual(result, expected)
     
     def test_eval_reference(self):
         v = self.client[self.client.ids[0]]
         v['g'] = range(5)
         rg = pmod.Reference('g[0]')
         echo = lambda x:x
         self.assertEqual(v.apply_sync(echo, rg), 0)
     
     def test_reference_nameerror(self):
         v = self.client[self.client.ids[0]]
         r = pmod.Reference('elvis_has_left')
         echo = lambda x:x
         self.assertRaisesRemote(NameError, v.apply_sync, echo, r)
 
     def test_single_engine_map(self):
         e0 = self.client[self.client.ids[0]]
         r = range(5)
         check = [ -1*i for i in r ]
         result = e0.map_sync(lambda x: -1*x, r)
         self.assertEqual(result, check)
     
     def test_len(self):
         """len(view) makes sense"""
         e0 = self.client[self.client.ids[0]]
         yield self.assertEqual(len(e0), 1)
         v = self.client[:]
         yield self.assertEqual(len(v), len(self.client.ids))
         v = self.client.direct_view('all')
         yield self.assertEqual(len(v), len(self.client.ids))
         v = self.client[:2]
         yield self.assertEqual(len(v), 2)
         v = self.client[:1]
         yield self.assertEqual(len(v), 1)
         v = self.client.load_balanced_view()
         yield self.assertEqual(len(v), len(self.client.ids))
         # parametric tests seem to require manual closing?
         self.client.close()
 
     
     # begin execute tests
     
     def test_execute_reply(self):
         e0 = self.client[self.client.ids[0]]
         e0.block = True
         ar = e0.execute("5", silent=False)
         er = ar.get()
         self.assertEqual(str(er), "<ExecuteReply[%i]: 5>" % er.execution_count)
         self.assertEqual(er.pyout['data']['text/plain'], '5')
 
     def test_execute_reply_stdout(self):
         e0 = self.client[self.client.ids[0]]
         e0.block = True
         ar = e0.execute("print (5)", silent=False)
         er = ar.get()
         self.assertEqual(er.stdout.strip(), '5')
         
     def test_execute_pyout(self):
         """execute triggers pyout with silent=False"""
         view = self.client[:]
         ar = view.execute("5", silent=False, block=True)
         
         expected = [{'text/plain' : '5'}] * len(view)
         mimes = [ out['data'] for out in ar.pyout ]
         self.assertEqual(mimes, expected)
     
     def test_execute_silent(self):
         """execute does not trigger pyout with silent=True"""
         view = self.client[:]
         ar = view.execute("5", block=True)
         expected = [None] * len(view)
         self.assertEqual(ar.pyout, expected)
     
     def test_execute_magic(self):
         """execute accepts IPython commands"""
         view = self.client[:]
         view.execute("a = 5")
         ar = view.execute("%whos", block=True)
         # this will raise, if that failed
         ar.get(5)
         for stdout in ar.stdout:
             lines = stdout.splitlines()
             self.assertEqual(lines[0].split(), ['Variable', 'Type', 'Data/Info'])
             found = False
             for line in lines[2:]:
                 split = line.split()
                 if split == ['a', 'int', '5']:
                     found = True
                     break
             self.assertTrue(found, "whos output wrong: %s" % stdout)
     
     def test_execute_displaypub(self):
         """execute tracks display_pub output"""
         view = self.client[:]
         view.execute("from IPython.core.display import *")
         ar = view.execute("[ display(i) for i in range(5) ]", block=True)
         
         expected = [ {u'text/plain' : unicode(j)} for j in range(5) ]
         for outputs in ar.outputs:
             mimes = [ out['data'] for out in outputs ]
             self.assertEqual(mimes, expected)
     
     def test_apply_displaypub(self):
         """apply tracks display_pub output"""
         view = self.client[:]
         view.execute("from IPython.core.display import *")
         
         @interactive
         def publish():
             [ display(i) for i in range(5) ]
         
         ar = view.apply_async(publish)
         ar.get(5)
         expected = [ {u'text/plain' : unicode(j)} for j in range(5) ]
         for outputs in ar.outputs:
             mimes = [ out['data'] for out in outputs ]
             self.assertEqual(mimes, expected)
     
     def test_execute_raises(self):
         """exceptions in execute requests raise appropriately"""
         view = self.client[-1]
         ar = view.execute("1/0")
         self.assertRaisesRemote(ZeroDivisionError, ar.get, 2)
     
     def test_remoteerror_render_exception(self):
         """RemoteErrors get nice tracebacks"""
         view = self.client[-1]
         ar = view.execute("1/0")
         ip = get_ipython()
         ip.user_ns['ar'] = ar
         with capture_output() as io:
             ip.run_cell("ar.get(2)")
         
         self.assertTrue('ZeroDivisionError' in io.stdout, io.stdout)
     
     def test_compositeerror_render_exception(self):
         """CompositeErrors get nice tracebacks"""
         view = self.client[:]
         ar = view.execute("1/0")
         ip = get_ipython()
         ip.user_ns['ar'] = ar
         with capture_output() as io:
             ip.run_cell("ar.get(2)")
         
         self.assertEqual(io.stdout.count('ZeroDivisionError'), len(view) * 2, io.stdout)
         self.assertEqual(io.stdout.count('by zero'), len(view), io.stdout)
         self.assertEqual(io.stdout.count(':execute'), len(view), io.stdout)
     
+    def test_compositeerror_truncate(self):
+        """Truncate CompositeErrors with many exceptions"""
+        view = self.client[:]
+        msg_ids = []
+        for i in range(10):
+            ar = view.execute("1/0")
+            msg_ids.extend(ar.msg_ids)
+        
+        ar = self.client.get_result(msg_ids)
+        try:
+            ar.get()
+        except error.CompositeError as e:
+            pass
+        else:
+            self.fail("Should have raised CompositeError")
+        
+        lines = e.render_traceback()
+        with capture_output() as io:
+            e.print_traceback()
+        
+        self.assertTrue("more exceptions" in lines[-1])
+        count = e.tb_limit
+        
+        self.assertEqual(io.stdout.count('ZeroDivisionError'), 2 * count, io.stdout)
+        self.assertEqual(io.stdout.count('by zero'), count, io.stdout)
+        self.assertEqual(io.stdout.count(':execute'), count, io.stdout)
+    
     @dec.skipif_not_matplotlib
     def test_magic_pylab(self):
         """%pylab works on engines"""
         view = self.client[-1]
         ar = view.execute("%pylab inline")
         # at least check if this raised:
         reply = ar.get(5)
         # include imports, in case user config
         ar = view.execute("plot(rand(100))", silent=False)
         reply = ar.get(5)
         self.assertEqual(len(reply.outputs), 1)
         output = reply.outputs[0]
         self.assertTrue("data" in output)
         data = output['data']
         self.assertTrue("image/png" in data)
     
     def test_func_default_func(self):
         """interactively defined function as apply func default"""
         def foo():
             return 'foo'
         
         def bar(f=foo):
             return f()
         
         view = self.client[-1]
         ar = view.apply_async(bar)
         r = ar.get(10)
         self.assertEqual(r, 'foo')
     def test_data_pub_single(self):
         view = self.client[-1]
         ar = view.execute('\n'.join([
             'from IPython.kernel.zmq.datapub import publish_data',
             'for i in range(5):',
             '  publish_data(dict(i=i))'
         ]), block=False)
         self.assertTrue(isinstance(ar.data, dict))
         ar.get(5)
         self.assertEqual(ar.data, dict(i=4))
 
     def test_data_pub(self):
         view = self.client[:]
         ar = view.execute('\n'.join([
             'from IPython.kernel.zmq.datapub import publish_data',
             'for i in range(5):',
             '  publish_data(dict(i=i))'
         ]), block=False)
         self.assertTrue(all(isinstance(d, dict) for d in ar.data))
         ar.get(5)
         self.assertEqual(ar.data, [dict(i=4)] * len(ar))
     
     def test_can_list_arg(self):
         """args in lists are canned"""
         view = self.client[-1]
         view['a'] = 128
         rA = pmod.Reference('a')
         ar = view.apply_async(lambda x: x, [rA])
         r = ar.get(5)
         self.assertEqual(r, [128])
 
     def test_can_dict_arg(self):
         """args in dicts are canned"""
         view = self.client[-1]
         view['a'] = 128
         rA = pmod.Reference('a')
         ar = view.apply_async(lambda x: x, dict(foo=rA))
         r = ar.get(5)
         self.assertEqual(r, dict(foo=128))
 
     def test_can_list_kwarg(self):
         """kwargs in lists are canned"""
         view = self.client[-1]
         view['a'] = 128
         rA = pmod.Reference('a')
         ar = view.apply_async(lambda x=5: x, x=[rA])
         r = ar.get(5)
         self.assertEqual(r, [128])
 
     def test_can_dict_kwarg(self):
         """kwargs in dicts are canned"""
         view = self.client[-1]
         view['a'] = 128
         rA = pmod.Reference('a')
         ar = view.apply_async(lambda x=5: x, dict(foo=rA))
         r = ar.get(5)
         self.assertEqual(r, dict(foo=128))
     
     def test_map_ref(self):
         """view.map works with references"""
         view = self.client[:]
         ranks = sorted(self.client.ids)
         view.scatter('rank', ranks, flatten=True)
         rrank = pmod.Reference('rank')
         
         amr = view.map_async(lambda x: x*2, [rrank] * len(view))
         drank = amr.get(5)
         self.assertEqual(drank, [ r*2 for r in ranks ])
         
     def test_nested_getitem_setitem(self):
         """get and set with view['a.b']"""
         view = self.client[-1]
         view.execute('\n'.join([
             'class A(object): pass',
             'a = A()',
             'a.b = 128',
             ]), block=True)
         ra = pmod.Reference('a')
 
         r = view.apply_sync(lambda x: x.b, ra)
         self.assertEqual(r, 128)
         self.assertEqual(view['a.b'], 128)
 
         view['a.b'] = 0
 
         r = view.apply_sync(lambda x: x.b, ra)
         self.assertEqual(r, 0)
         self.assertEqual(view['a.b'], 0)
     
     def test_return_namedtuple(self):
         def namedtuplify(x, y):
             from IPython.parallel.tests.test_view import point
             return point(x, y)
         
         view = self.client[-1]
         p = view.apply_sync(namedtuplify, 1, 2)
         self.assertEqual(p.x, 1)
         self.assertEqual(p.y, 2)
 
     def test_apply_namedtuple(self):
         def echoxy(p):
             return p.y, p.x
         
         view = self.client[-1]
         tup = view.apply_sync(echoxy, point(1, 2))
         self.assertEqual(tup, (2,1))
 

docs/source/parallel/parallel_multiengine.txt

@@ -1,694 +1,697 b''
 .. _parallel_multiengine:
 
 ==========================
 IPython's Direct interface
 ==========================
 
 The direct, or multiengine, interface represents one possible way of working with a set of
 IPython engines. The basic idea behind the multiengine interface is that the
 capabilities of each engine are directly and explicitly exposed to the user.
 Thus, in the multiengine interface, each engine is given an id that is used to
 identify the engine and give it work to do. This interface is very intuitive
 and is designed with interactive usage in mind, and is the best place for
 new users of IPython to begin.
 
 Starting the IPython controller and engines
 ===========================================
 
 To follow along with this tutorial, you will need to start the IPython
 controller and four IPython engines. The simplest way of doing this is to use
 the :command:`ipcluster` command::
 
     $ ipcluster start -n 4
     
 For more detailed information about starting the controller and engines, see
 our :ref:`introduction <parallel_overview>` to using IPython for parallel computing.
 
 Creating a ``DirectView`` instance
 ==================================
 
 The first step is to import the IPython :mod:`IPython.parallel`
 module and then create a :class:`.Client` instance:
 
 .. sourcecode:: ipython
 
     In [1]: from IPython.parallel import Client
     
     In [2]: rc = Client()
 
 This form assumes that the default connection information (stored in
 :file:`ipcontroller-client.json` found in :file:`IPYTHONDIR/profile_default/security`) is
 accurate. If the controller was started on a remote machine, you must copy that connection
 file to the client machine, or enter its contents as arguments to the Client constructor:
 
 .. sourcecode:: ipython
 
     # If you have copied the json connector file from the controller:
     In [2]: rc = Client('/path/to/ipcontroller-client.json')
     # or to connect with a specific profile you have set up:
     In [3]: rc = Client(profile='mpi')
     
 
 To make sure there are engines connected to the controller, users can get a list
 of engine ids:
 
 .. sourcecode:: ipython
 
     In [3]: rc.ids
     Out[3]: [0, 1, 2, 3]
 
 Here we see that there are four engines ready to do work for us.
 
 For direct execution, we will make use of a :class:`DirectView` object, which can be
 constructed via list-access to the client:
 
 .. sourcecode:: ipython
 
     In [4]: dview = rc[:] # use all engines
 
 .. seealso::
 
     For more information, see the in-depth explanation of :ref:`Views <parallel_details>`.
 
 
 Quick and easy parallelism
 ==========================
 
 In many cases, you simply want to apply a Python function to a sequence of
 objects, but *in parallel*. The client interface provides a simple way
 of accomplishing this: using the DirectView's :meth:`~DirectView.map` method.
 
 Parallel map
 ------------
 
 Python's builtin :func:`map` functions allows a function to be applied to a
 sequence element-by-element. This type of code is typically trivial to
 parallelize. In fact, since IPython's interface is all about functions anyway,
 you can just use the builtin :func:`map` with a :class:`RemoteFunction`, or a 
 DirectView's :meth:`map` method:
 
 .. sourcecode:: ipython
 
     In [62]: serial_result = map(lambda x:x**10, range(32))
     
     In [63]: parallel_result = dview.map_sync(lambda x: x**10, range(32))
 
     In [67]: serial_result==parallel_result
     Out[67]: True
 
 
 .. note::
 
     The :class:`DirectView`'s version of :meth:`map` does
     not do dynamic load balancing. For a load balanced version, use a
     :class:`LoadBalancedView`.
 
 .. seealso::
     
     :meth:`map` is implemented via :class:`ParallelFunction`.
 
 Remote function decorators
 --------------------------
 
 Remote functions are just like normal functions, but when they are called,
 they execute on one or more engines, rather than locally. IPython provides
 two decorators:
 
 .. sourcecode:: ipython
 
     In [10]: @dview.remote(block=True)
        ....: def getpid():
        ....:     import os
        ....:     return os.getpid()
        ....: 
 
     In [11]: getpid()
     Out[11]: [12345, 12346, 12347, 12348]
 
 The ``@parallel`` decorator creates parallel functions, that break up an element-wise
 operations and distribute them, reconstructing the result.
 
 .. sourcecode:: ipython
 
     In [12]: import numpy as np
     
     In [13]: A = np.random.random((64,48))
     
     In [14]: @dview.parallel(block=True)
        ....: def pmul(A,B):
        ....:     return A*B
     
     In [15]: C_local = A*A
     
     In [16]: C_remote = pmul(A,A)
     
     In [17]: (C_local == C_remote).all()
     Out[17]: True
 
 Calling a ``@parallel`` function *does not* correspond to map. It is used for splitting
 element-wise operations that operate on a sequence or array.  For ``map`` behavior,
 parallel functions do have a map method.
 
 ====================    ============================    =============================
 call                    pfunc(seq)                      pfunc.map(seq)
 ====================    ============================    =============================
 # of tasks              # of engines (1 per engine)     # of engines (1 per engine)
 # of remote calls       # of engines (1 per engine)     ``len(seq)``
 argument to remote      ``seq[i:j]`` (sub-sequence)     ``seq[i]`` (single element)
 ====================    ============================    =============================
 
 A quick example to illustrate the difference in arguments for the two modes:
 
 .. sourcecode:: ipython
 
     In [16]: @dview.parallel(block=True)
        ....: def echo(x):
        ....:     return str(x)
        ....: 
 
     In [17]: echo(range(5))
     Out[17]: ['[0, 1]', '[2]', '[3]', '[4]']
 
     In [18]: echo.map(range(5))
     Out[18]: ['0', '1', '2', '3', '4']
 
 
 .. seealso::
 
     See the :func:`~.remotefunction.parallel` and :func:`~.remotefunction.remote`
     decorators for options.
 
 Calling Python functions
 ========================
 
 The most basic type of operation that can be performed on the engines is to
 execute Python code or call Python functions. Executing Python code can be
 done in blocking or non-blocking mode (non-blocking is default) using the
 :meth:`.View.execute` method, and calling functions can be done via the
 :meth:`.View.apply` method.
 
 apply
 -----
 
 The main method for doing remote execution (in fact, all methods that
 communicate with the engines are built on top of it), is :meth:`View.apply`.
 
 We strive to provide the cleanest interface we can, so `apply` has the following
 signature:
 
 .. sourcecode:: python
 
     view.apply(f, *args, **kwargs)
 
 There are various ways to call functions with IPython, and these flags are set as
 attributes of the View.  The ``DirectView`` has just two of these flags:
 
 dv.block : bool
     whether to wait for the result, or return an :class:`AsyncResult` object
     immediately
 dv.track : bool
     whether to instruct pyzmq to track when zeromq is done sending the message.
     This is primarily useful for non-copying sends of numpy arrays that you plan to
     edit in-place.  You need to know when it becomes safe to edit the buffer
     without corrupting the message.
 dv.targets : int, list of ints
     which targets this view is associated with.
 
 
 Creating a view is simple: index-access on a client creates a :class:`.DirectView`.
 
 .. sourcecode:: ipython
 
     In [4]: view = rc[1:3]
     Out[4]: <DirectView [1, 2]>
 
     In [5]: view.apply<tab>
     view.apply  view.apply_async  view.apply_sync
 
 For convenience, you can set block temporarily for a single call with the extra sync/async methods.
 
 Blocking execution
 ------------------
 
 In blocking mode, the :class:`.DirectView` object (called ``dview`` in
 these examples) submits the command to the controller, which places the
 command in the engines' queues for execution. The :meth:`apply` call then
 blocks until the engines are done executing the command:
 
 .. sourcecode:: ipython
 
     In [2]: dview = rc[:] # A DirectView of all engines
     In [3]: dview.block=True
     In [4]: dview['a'] = 5
 
     In [5]: dview['b'] = 10
 
     In [6]: dview.apply(lambda x: a+b+x, 27)
     Out[6]: [42, 42, 42, 42]
 
 You can also select blocking execution on a call-by-call basis with the :meth:`apply_sync`
 method:
 
 .. sourcecode:: ipython
 
     In [7]: dview.block=False
 
     In [8]: dview.apply_sync(lambda x: a+b+x, 27)
     Out[8]: [42, 42, 42, 42]
 
 Python commands can be executed as strings on specific engines by using a View's ``execute``
 method:
 
 .. sourcecode:: ipython
 
     In [6]: rc[::2].execute('c=a+b')
 
     In [7]: rc[1::2].execute('c=a-b')
 
     In [8]: dview['c'] # shorthand for dview.pull('c', block=True)
     Out[8]: [15, -5, 15, -5]
 
 
 Non-blocking execution
 ----------------------
 
 In non-blocking mode, :meth:`apply` submits the command to be executed and
 then returns a :class:`AsyncResult` object immediately. The
 :class:`AsyncResult` object gives you a way of getting a result at a later
 time through its :meth:`get` method.
 
 .. seealso::
 
     Docs on the :ref:`AsyncResult <parallel_asyncresult>` object.
 
 This allows you to quickly submit long running commands without blocking your
 local Python/IPython session:
 
 .. sourcecode:: ipython
     
     # define our function
     In [6]: def wait(t):
       ....:     import time
       ....:     tic = time.time()
       ....:     time.sleep(t)
       ....:     return time.time()-tic
     
     # In non-blocking mode
     In [7]: ar = dview.apply_async(wait, 2)
 
     # Now block for the result
     In [8]: ar.get()
     Out[8]: [2.0006198883056641, 1.9997570514678955, 1.9996809959411621, 2.0003249645233154]
 
     # Again in non-blocking mode
     In [9]: ar = dview.apply_async(wait, 10)
 
     # Poll to see if the result is ready
     In [10]: ar.ready()
     Out[10]: False
     
     # ask for the result, but wait a maximum of 1 second:
     In [45]: ar.get(1)
     ---------------------------------------------------------------------------
     TimeoutError                              Traceback (most recent call last)
     /home/you/<ipython-input-45-7cd858bbb8e0> in <module>()
     ----> 1 ar.get(1)
 
     /path/to/site-packages/IPython/parallel/asyncresult.pyc in get(self, timeout)
          62                 raise self._exception
          63         else:
     ---> 64             raise error.TimeoutError("Result not ready.")
          65 
          66     def ready(self):
 
     TimeoutError: Result not ready.
 
 .. Note::
 
     Note the import inside the function. This is a common model, to ensure
     that the appropriate modules are imported where the task is run. You can
     also manually import modules into the engine(s) namespace(s) via 
     :meth:`view.execute('import numpy')`.
 
 Often, it is desirable to wait until a set of :class:`AsyncResult` objects
 are done. For this, there is a the method :meth:`wait`. This method takes a
 tuple of :class:`AsyncResult` objects (or `msg_ids` or indices to the client's History),
 and blocks until all of the associated results are ready:
 
 .. sourcecode:: ipython
 
     In [72]: dview.block=False
 
     # A trivial list of AsyncResults objects
     In [73]: pr_list = [dview.apply_async(wait, 3) for i in range(10)]
 
     # Wait until all of them are done
     In [74]: dview.wait(pr_list)
 
     # Then, their results are ready using get() or the `.r` attribute
     In [75]: pr_list[0].get()
     Out[75]: [2.9982571601867676, 2.9982588291168213, 2.9987530708312988, 2.9990990161895752]
 
 
 
 The ``block`` and ``targets`` keyword arguments and attributes
 --------------------------------------------------------------
 
 Most DirectView methods (excluding :meth:`apply`) accept ``block`` and
 ``targets`` as keyword arguments. As we have seen above, these keyword arguments control the
 blocking mode and which engines the command is applied to. The :class:`View` class also has
 :attr:`block` and :attr:`targets` attributes that control the default behavior when the keyword
 arguments are not provided. Thus the following logic is used for :attr:`block` and :attr:`targets`:
 
 * If no keyword argument is provided, the instance attributes are used.
 * The Keyword arguments, if provided overrides the instance attributes for
   the duration of a single call.
   
 The following examples demonstrate how to use the instance attributes:
 
 .. sourcecode:: ipython
 
     In [16]: dview.targets = [0,2]
     
     In [17]: dview.block = False
 
     In [18]: ar = dview.apply(lambda : 10)
 
     In [19]: ar.get()
     Out[19]: [10, 10]
 
     In [20]: dview.targets = v.client.ids # all engines (4)
     
     In [21]: dview.block = True
 
     In [22]: dview.apply(lambda : 42)
     Out[22]: [42, 42, 42, 42]
 
 The :attr:`block` and :attr:`targets` instance attributes of the
 :class:`.DirectView` also determine the behavior of the parallel magic commands.
 
 .. seealso::
 
     See the documentation of the :ref:`Parallel Magics <parallel_magics>`.
 
 
 Moving Python objects around
 ============================
 
 In addition to calling functions and executing code on engines, you can
 transfer Python objects to and from your IPython session and the engines. In
 IPython, these operations are called :meth:`push` (sending an object to the
 engines) and :meth:`pull` (getting an object from the engines).
 
 Basic push and pull
 -------------------
 
 Here are some examples of how you use :meth:`push` and :meth:`pull`:
 
 .. sourcecode:: ipython
 
     In [38]: dview.push(dict(a=1.03234,b=3453))
     Out[38]: [None,None,None,None]
 
     In [39]: dview.pull('a')
     Out[39]: [ 1.03234, 1.03234, 1.03234, 1.03234]
 
     In [40]: dview.pull('b', targets=0)
     Out[40]: 3453
 
     In [41]: dview.pull(('a','b'))
     Out[41]: [ [1.03234, 3453], [1.03234, 3453], [1.03234, 3453], [1.03234, 3453] ]
     
     In [42]: dview.push(dict(c='speed'))
     Out[42]: [None,None,None,None]
 
 In non-blocking mode :meth:`push` and :meth:`pull` also return
 :class:`AsyncResult` objects:
 
 .. sourcecode:: ipython
 
     In [48]: ar = dview.pull('a', block=False)
 
     In [49]: ar.get()
     Out[49]: [1.03234, 1.03234, 1.03234, 1.03234]
 
 
 Dictionary interface
 --------------------
 
 Since a Python namespace is just a :class:`dict`, :class:`DirectView` objects provide
 dictionary-style access by key and methods such as :meth:`get` and
 :meth:`update` for convenience. This make the remote namespaces of the engines
 appear as a local dictionary. Underneath, these methods call :meth:`apply`:
 
 .. sourcecode:: ipython
 
     In [51]: dview['a']=['foo','bar']
 
     In [52]: dview['a']
     Out[52]: [ ['foo', 'bar'], ['foo', 'bar'], ['foo', 'bar'], ['foo', 'bar'] ]
 
 Scatter and gather
 ------------------
 
 Sometimes it is useful to partition a sequence and push the partitions to
 different engines. In MPI language, this is know as scatter/gather and we
 follow that terminology. However, it is important to remember that in
 IPython's :class:`Client` class, :meth:`scatter` is from the
 interactive IPython session to the engines and :meth:`gather` is from the
 engines back to the interactive IPython session. For scatter/gather operations
 between engines, MPI, pyzmq, or some other direct interconnect should be used.
 
 .. sourcecode:: ipython
 
     In [58]: dview.scatter('a',range(16))
     Out[58]: [None,None,None,None]
 
     In [59]: dview['a']
     Out[59]: [ [0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13, 14, 15] ]
 
     In [60]: dview.gather('a')
     Out[60]: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]
 
 Other things to look at
 =======================
 
 How to do parallel list comprehensions
 --------------------------------------
 
 In many cases list comprehensions are nicer than using the map function. While
 we don't have fully parallel list comprehensions, it is simple to get the
 basic effect using :meth:`scatter` and :meth:`gather`:
 
 .. sourcecode:: ipython
 
     In [66]: dview.scatter('x',range(64))
 
     In [67]: %px y = [i**10 for i in x]
     Parallel execution on engines: [0, 1, 2, 3]
 
     In [68]: y = dview.gather('y')
 
     In [69]: print y
     [0, 1, 1024, 59049, 1048576, 9765625, 60466176, 282475249, 1073741824,...]
 
 Remote imports
 --------------
 
 Sometimes you will want to import packages both in your interactive session
 and on your remote engines.  This can be done with the :class:`ContextManager`
 created by a DirectView's :meth:`sync_imports` method:
 
 .. sourcecode:: ipython
 
     In [69]: with dview.sync_imports():
        ....:     import numpy
     importing numpy on engine(s)
 
 Any imports made inside the block will also be performed on the view's engines.
 sync_imports also takes a `local` boolean flag that defaults to True, which specifies
 whether the local imports should also be performed.  However, support for `local=False`
 has not been implemented, so only packages that can be imported locally will work
 this way.
 
 You can also specify imports via the ``@require`` decorator.  This is a decorator
 designed for use in Dependencies, but can be used to handle remote imports as well.
 Modules or module names passed to ``@require`` will be imported before the decorated
 function is called.  If they cannot be imported, the decorated function will never
 execute and will fail with an UnmetDependencyError. Failures of single Engines will
 be collected and raise a CompositeError, as demonstrated in the next section.
 
 .. sourcecode:: ipython
 
     In [69]: from IPython.parallel import require
 
     In [70]: @require('re'):
        ....: def findall(pat, x):
        ....:     # re is guaranteed to be available
        ....:     return re.findall(pat, x)
           
     # you can also pass modules themselves, that you already have locally:
     In [71]: @require(time):
        ....: def wait(t):
        ....:     time.sleep(t)
        ....:     return t
 
 .. note::
 
     :func:`sync_imports` does not allow ``import foo as bar`` syntax,
     because the assignment represented by the ``as bar`` part is not
     available to the import hook.
 
 
 .. _parallel_exceptions:
 
 Parallel exceptions
 -------------------
 
 In the multiengine interface, parallel commands can raise Python exceptions,
 just like serial commands. But it is a little subtle, because a single
 parallel command can actually raise multiple exceptions (one for each engine
 the command was run on). To express this idea, we have a
 :exc:`CompositeError` exception class that will be raised in most cases. The
 :exc:`CompositeError` class is a special type of exception that wraps one or
 more other types of exceptions. Here is how it works:
 
 .. sourcecode:: ipython
 
     In [78]: dview.block = True
     
     In [79]: dview.execute("1/0")
     [0:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
     [1:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
     [2:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
     [3:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
 Notice how the error message printed when :exc:`CompositeError` is raised has
 information about the individual exceptions that were raised on each engine.
 If you want, you can even raise one of these original exceptions:
 
 .. sourcecode:: ipython
 
     In [80]: try:
        ....:     dview.execute('1/0', block=True)
        ....: except parallel.error.CompositeError, e:
        ....:     e.raise_exception()
        ....: 
        ....: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
 If you are working in IPython, you can simple type ``%debug`` after one of
 these :exc:`CompositeError` exceptions is raised, and inspect the exception
 instance:
 
 .. sourcecode:: ipython
 
     In [81]: dview.execute('1/0')
     [0:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
     [1:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
     [2:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
     [3:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
     
     In [82]: %debug
     > /.../site-packages/IPython/parallel/client/asyncresult.py(125)get()
         124             else:
     --> 125                 raise self._exception
         126         else:
     
     # Here, self._exception is the CompositeError instance:
     
     ipdb> e = self._exception
     ipdb> e
     CompositeError(4)
     
     # we can tab-complete on e to see available methods:
     ipdb> e.<TAB>
     e.args               e.message            e.traceback
     e.elist              e.msg
     e.ename              e.print_traceback
     e.engine_info        e.raise_exception
     e.evalue             e.render_traceback
     
     # We can then display the individual tracebacks, if we want:
     ipdb> e.print_traceback(1)
     [1:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
 
 
+Since you might have 100 engines, you probably don't want to see 100 tracebacks
+for a simple NameError because of a typo.
+For this reason, CompositeError truncates the list of exceptions it will print
+to :attr:`CompositeError.tb_limit` (default is five).
+You can change this limit to suit your needs with:
+
+.. sourcecode:: ipython
+
+    In [20]: from IPython.parallel import CompositeError
+    In [21]: CompositeError.tb_limit = 1
+    In [22]: %px a=b
+    [0:execute]: 
+    ---------------------------------------------------------------------------
+    NameError                                 Traceback (most recent call last)
+    ----> 1 a=b
+    NameError: name 'b' is not defined
+
+    ... 3 more exceptions ...
+
+
 All of this same error handling magic even works in non-blocking mode:
 
 .. sourcecode:: ipython
 
     In [83]: dview.block=False
 
     In [84]: ar = dview.execute('1/0')
 
     In [85]: ar.get()
     [0:execute]: 
     ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
-    ----> 1 1/0
-    ZeroDivisionError: integer division or modulo by zero
-
-    [1:execute]: 
-    ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
+    ZeroDivisionError                         Traceback (most recent call last)
     ----> 1 1/0
     ZeroDivisionError: integer division or modulo by zero
-
-    [2:execute]: 
-    ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
-    ----> 1 1/0
-    ZeroDivisionError: integer division or modulo by zero
-
-    [3:execute]: 
-    ---------------------------------------------------------------------------
-    ZeroDivisionError                         Traceback (most recent call last)<ipython-input-1-05c9758a9c21> in <module>()
-    ----> 1 1/0
-    ZeroDivisionError: integer division or modulo by zero
-
+    
+    ... 3 more exceptions ...