remove some extraneous print statements from IPython.parallel

closes  #4203

backport for 1.2

IPython/parallel/client/remotefunction.py

@@ -1,283 +1,282 b''
 """Remote Functions and decorators for Views.
 
 Authors:
 
 * Brian Granger
 * Min RK
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-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
 #-----------------------------------------------------------------------------
 
 from __future__ import division
 
 import sys
 import warnings
 
 from IPython.external.decorator import decorator
 from IPython.testing.skipdoctest import skip_doctest
 
 from . import map as Map
 from .asyncresult import AsyncMapResult
 
 #-----------------------------------------------------------------------------
 # Functions and Decorators
 #-----------------------------------------------------------------------------
 
 @skip_doctest
 def remote(view, block=None, **flags):
     """Turn a function into a remote function.
 
     This method can be used for map:
 
     In [1]: @remote(view,block=True)
        ...: def func(a):
        ...:    pass
     """
 
     def remote_function(f):
         return RemoteFunction(view, f, block=block, **flags)
     return remote_function
 
 @skip_doctest
 def parallel(view, dist='b', block=None, ordered=True, **flags):
     """Turn a function into a parallel remote function.
 
     This method can be used for map:
 
     In [1]: @parallel(view, block=True)
        ...: def func(a):
        ...:    pass
     """
 
     def parallel_function(f):
         return ParallelFunction(view, f, dist=dist, block=block, ordered=ordered, **flags)
     return parallel_function
 
 def getname(f):
     """Get the name of an object.
     
     For use in case of callables that are not functions, and
     thus may not have __name__ defined.
     
     Order: f.__name__ >  f.name > str(f)
     """
     try:
         return f.__name__
     except:
         pass
     try:
         return f.name
     except:
         pass
     
     return str(f)
 
 @decorator
 def sync_view_results(f, self, *args, **kwargs):
     """sync relevant results from self.client to our results attribute.
     
     This is a clone of view.sync_results, but for remote functions
     """
     view = self.view
     if view._in_sync_results:
         return f(self, *args, **kwargs)
-    print 'in sync results', f
     view._in_sync_results = True
     try:
         ret = f(self, *args, **kwargs)
     finally:
         view._in_sync_results = False
         view._sync_results()
     return ret
     
 #--------------------------------------------------------------------------
 # Classes
 #--------------------------------------------------------------------------
 
 class RemoteFunction(object):
     """Turn an existing function into a remote function.
 
     Parameters
     ----------
 
     view : View instance
         The view to be used for execution
     f : callable
         The function to be wrapped into a remote function
     block : bool [default: None]
         Whether to wait for results or not.  The default behavior is
         to use the current `block` attribute of `view`
 
     **flags : remaining kwargs are passed to View.temp_flags
     """
 
     view = None # the remote connection
     func = None # the wrapped function
     block = None # whether to block
     flags = None # dict of extra kwargs for temp_flags
 
     def __init__(self, view, f, block=None, **flags):
         self.view = view
         self.func = f
         self.block=block
         self.flags=flags
 
     def __call__(self, *args, **kwargs):
         block = self.view.block if self.block is None else self.block
         with self.view.temp_flags(block=block, **self.flags):
             return self.view.apply(self.func, *args, **kwargs)
 
 
 class ParallelFunction(RemoteFunction):
     """Class for mapping a function to sequences.
 
     This will distribute the sequences according the a mapper, and call
     the function on each sub-sequence.  If called via map, then the function
     will be called once on each element, rather that each sub-sequence.
 
     Parameters
     ----------
 
     view : View instance
         The view to be used for execution
     f : callable
         The function to be wrapped into a remote function
     dist : str [default: 'b']
         The key for which mapObject to use to distribute sequences
         options are:
           * 'b' : use contiguous chunks in order
           * 'r' : use round-robin striping
     block : bool [default: None]
         Whether to wait for results or not.  The default behavior is
         to use the current `block` attribute of `view`
     chunksize : int or None
         The size of chunk to use when breaking up sequences in a load-balanced manner
     ordered : bool [default: True]
         Whether the result should be kept in order. If False,
         results become available as they arrive, regardless of submission order.
     **flags : remaining kwargs are passed to View.temp_flags
     """
 
     chunksize = None
     ordered = None
     mapObject = None
     _mapping = False
 
     def __init__(self, view, f, dist='b', block=None, chunksize=None, ordered=True, **flags):
         super(ParallelFunction, self).__init__(view, f, block=block, **flags)
         self.chunksize = chunksize
         self.ordered = ordered
 
         mapClass = Map.dists[dist]
         self.mapObject = mapClass()
     
     @sync_view_results
     def __call__(self, *sequences):
         client = self.view.client
         
         lens = []
         maxlen = minlen = -1
         for i, seq in enumerate(sequences):
             try:
                 n = len(seq)
             except Exception:
                 seq = list(seq)
                 if isinstance(sequences, tuple):
                     # can't alter a tuple
                     sequences = list(sequences)
                 sequences[i] = seq
                 n = len(seq)
             if n > maxlen:
                 maxlen = n
             if minlen == -1 or n < minlen:
                 minlen = n
             lens.append(n)
         
         # check that the length of sequences match
         if not self._mapping and minlen != maxlen:
             msg = 'all sequences must have equal length, but have %s' % lens
             raise ValueError(msg)
         
         balanced = 'Balanced' in self.view.__class__.__name__
         if balanced:
             if self.chunksize:
                 nparts = maxlen // self.chunksize + int(maxlen % self.chunksize > 0)
             else:
                 nparts = maxlen
             targets = [None]*nparts
         else:
             if self.chunksize:
                 warnings.warn("`chunksize` is ignored unless load balancing", UserWarning)
             # multiplexed:
             targets = self.view.targets
             # 'all' is lazily evaluated at execution time, which is now:
             if targets == 'all':
                 targets = client._build_targets(targets)[1]
             elif isinstance(targets, int):
                 # single-engine view, targets must be iterable
                 targets = [targets]
             nparts = len(targets)
 
         msg_ids = []
         for index, t in enumerate(targets):
             args = []
             for seq in sequences:
                 part = self.mapObject.getPartition(seq, index, nparts, maxlen)
                 args.append(part)
 
             if sum([len(arg) for arg in args]) == 0:
                 continue
 
             if self._mapping:
                 if sys.version_info[0] >= 3:
                     f = lambda f, *sequences: list(map(f, *sequences))
                 else:
                     f = map
                 args = [self.func] + args
             else:
                 f=self.func
 
             view = self.view if balanced else client[t]
             with view.temp_flags(block=False, **self.flags):
                 ar = view.apply(f, *args)
 
             msg_ids.extend(ar.msg_ids)
 
         r = AsyncMapResult(self.view.client, msg_ids, self.mapObject,
                             fname=getname(self.func),
                             ordered=self.ordered
                         )
 
         if self.block:
             try:
                 return r.get()
             except KeyboardInterrupt:
                 return r
         else:
             return r
 
     def map(self, *sequences):
         """call a function on each element of one or more sequence(s) remotely.
         This should behave very much like the builtin map, but return an AsyncMapResult
         if self.block is False.
         
         That means it can take generators (will be cast to lists locally),
         and mismatched sequence lengths will be padded with None.
         """
         # set _mapping as a flag for use inside self.__call__
         self._mapping = True
         try:
             ret = self(*sequences)
         finally:
             self._mapping = False
         return ret
 
 __all__ = ['remote', 'parallel', 'RemoteFunction', 'ParallelFunction']

IPython/parallel/tests/__init__.py

@@ -1,126 +1,125 b''
 """toplevel setup/teardown for parallel tests."""
 
 #-------------------------------------------------------------------------------
 #  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 os
 import tempfile
 import time
 from subprocess import Popen
 
 from IPython.utils.path import get_ipython_dir
 from IPython.parallel import Client
 from IPython.parallel.apps.launcher import (LocalProcessLauncher,
                                                   ipengine_cmd_argv,
                                                   ipcontroller_cmd_argv,
                                                   SIGKILL,
                                                   ProcessStateError,
 )
 
 # globals
 launchers = []
 blackhole = open(os.devnull, 'w')
 
 # Launcher class
 class TestProcessLauncher(LocalProcessLauncher):
     """subclass LocalProcessLauncher, to prevent extra sockets and threads being created on Windows"""
     def start(self):
         if self.state == 'before':
             self.process = Popen(self.args,
                 stdout=blackhole, stderr=blackhole,
                 env=os.environ,
                 cwd=self.work_dir
             )
             self.notify_start(self.process.pid)
             self.poll = self.process.poll
         else:
             s = 'The process was already started and has state: %r' % self.state
             raise ProcessStateError(s)
 
 # nose setup/teardown
 
 def setup():
     cluster_dir = os.path.join(get_ipython_dir(), 'profile_iptest')
     engine_json = os.path.join(cluster_dir, 'security', 'ipcontroller-engine.json')
     client_json = os.path.join(cluster_dir, 'security', 'ipcontroller-client.json')
     for json in (engine_json, client_json):
         if os.path.exists(json):
             os.remove(json)
     
     cp = TestProcessLauncher()
     cp.cmd_and_args = ipcontroller_cmd_argv + \
                 ['--profile=iptest', '--log-level=50', '--ping=250', '--dictdb']
     cp.start()
     launchers.append(cp)
     tic = time.time()
     while not os.path.exists(engine_json) or not os.path.exists(client_json):
         if cp.poll() is not None:
-            print cp.poll()
-            raise RuntimeError("The test controller failed to start.")
+            raise RuntimeError("The test controller exited with status %s" % cp.poll())
         elif time.time()-tic > 15:
             raise RuntimeError("Timeout waiting for the test controller to start.")
         time.sleep(0.1)
     add_engines(1)
 
 def add_engines(n=1, profile='iptest', total=False):
     """add a number of engines to a given profile.
     
     If total is True, then already running engines are counted, and only
     the additional engines necessary (if any) are started.
     """
     rc = Client(profile=profile)
     base = len(rc)
     
     if total:
         n = max(n - base, 0)
     
     eps = []
     for i in range(n):
         ep = TestProcessLauncher()
         ep.cmd_and_args = ipengine_cmd_argv + [
             '--profile=%s' % profile,
             '--log-level=50',
             '--InteractiveShell.colors=nocolor'
             ]
         ep.start()
         launchers.append(ep)
         eps.append(ep)
     tic = time.time()
     while len(rc) < base+n:
         if any([ ep.poll() is not None for ep in eps ]):
             raise RuntimeError("A test engine failed to start.")
         elif time.time()-tic > 15:
             raise RuntimeError("Timeout waiting for engines to connect.")
         time.sleep(.1)
         rc.spin()
     rc.close()
     return eps
 
 def teardown():
     time.sleep(1)
     while launchers:
         p = launchers.pop()
         if p.poll() is None:
             try:
                 p.stop()
             except Exception as e:
                 print e
                 pass
         if p.poll() is None:
             time.sleep(.25)
         if p.poll() is None:
             try:
                 print 'cleaning up test process...'
                 p.signal(SIGKILL)
             except:
                 print "couldn't shutdown process: ", p
     blackhole.close()
     

IPython/parallel/tests/test_client.py

@@ -1,518 +1,517 b''
 """Tests for parallel client.py
 
 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
 #-------------------------------------------------------------------------------
 
 from __future__ import division
 
 import time
 from datetime import datetime
 from tempfile import mktemp
 
 import zmq
 
 from IPython import parallel
 from IPython.parallel.client import client as clientmod
 from IPython.parallel import error
 from IPython.parallel import AsyncResult, AsyncHubResult
 from IPython.parallel import LoadBalancedView, DirectView
 
 from clienttest import ClusterTestCase, segfault, wait, add_engines
 
 def setup():
     add_engines(4, total=True)
 
 class TestClient(ClusterTestCase):
     
     def test_ids(self):
         n = len(self.client.ids)
         self.add_engines(2)
         self.assertEqual(len(self.client.ids), n+2)
     
     def test_view_indexing(self):
         """test index access for views"""
         self.minimum_engines(4)
         targets = self.client._build_targets('all')[-1]
         v = self.client[:]
         self.assertEqual(v.targets, targets)
         t = self.client.ids[2]
         v = self.client[t]
         self.assertTrue(isinstance(v, DirectView))
         self.assertEqual(v.targets, t)
         t = self.client.ids[2:4]
         v = self.client[t]
         self.assertTrue(isinstance(v, DirectView))
         self.assertEqual(v.targets, t)
         v = self.client[::2]
         self.assertTrue(isinstance(v, DirectView))
         self.assertEqual(v.targets, targets[::2])
         v = self.client[1::3]
         self.assertTrue(isinstance(v, DirectView))
         self.assertEqual(v.targets, targets[1::3])
         v = self.client[:-3]
         self.assertTrue(isinstance(v, DirectView))
         self.assertEqual(v.targets, targets[:-3])
         v = self.client[-1]
         self.assertTrue(isinstance(v, DirectView))
         self.assertEqual(v.targets, targets[-1])
         self.assertRaises(TypeError, lambda : self.client[None])
     
     def test_lbview_targets(self):
         """test load_balanced_view targets"""
         v = self.client.load_balanced_view()
         self.assertEqual(v.targets, None)
         v = self.client.load_balanced_view(-1)
         self.assertEqual(v.targets, [self.client.ids[-1]])
         v = self.client.load_balanced_view('all')
         self.assertEqual(v.targets, None)
     
     def test_dview_targets(self):
         """test direct_view targets"""
         v = self.client.direct_view()
         self.assertEqual(v.targets, 'all')
         v = self.client.direct_view('all')
         self.assertEqual(v.targets, 'all')
         v = self.client.direct_view(-1)
         self.assertEqual(v.targets, self.client.ids[-1])
     
     def test_lazy_all_targets(self):
         """test lazy evaluation of rc.direct_view('all')"""
         v = self.client.direct_view()
         self.assertEqual(v.targets, 'all')
         
         def double(x):
             return x*2
         seq = range(100)
         ref = [ double(x) for x in seq ]
         
         # add some engines, which should be used
         self.add_engines(1)
         n1 = len(self.client.ids)
         
         # simple apply
         r = v.apply_sync(lambda : 1)
         self.assertEqual(r, [1] * n1)
         
         # map goes through remotefunction
         r = v.map_sync(double, seq)
         self.assertEqual(r, ref)
 
         # add a couple more engines, and try again
         self.add_engines(2)
         n2 = len(self.client.ids)
         self.assertNotEqual(n2, n1)
         
         # apply
         r = v.apply_sync(lambda : 1)
         self.assertEqual(r, [1] * n2)
         
         # map
         r = v.map_sync(double, seq)
         self.assertEqual(r, ref)
     
     def test_targets(self):
         """test various valid targets arguments"""
         build = self.client._build_targets
         ids = self.client.ids
         idents,targets = build(None)
         self.assertEqual(ids, targets)
     
     def test_clear(self):
         """test clear behavior"""
         self.minimum_engines(2)
         v = self.client[:]
         v.block=True
         v.push(dict(a=5))
         v.pull('a')
         id0 = self.client.ids[-1]
         self.client.clear(targets=id0, block=True)
         a = self.client[:-1].get('a')
         self.assertRaisesRemote(NameError, self.client[id0].get, 'a')
         self.client.clear(block=True)
         for i in self.client.ids:
             self.assertRaisesRemote(NameError, self.client[i].get, 'a')
     
     def test_get_result(self):
         """test getting results from the Hub."""
         c = clientmod.Client(profile='iptest')
         t = c.ids[-1]
         ar = c[t].apply_async(wait, 1)
         # give the monitor time to notice the message
         time.sleep(.25)
         ahr = self.client.get_result(ar.msg_ids[0])
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEqual(ahr.get(), ar.get())
         ar2 = self.client.get_result(ar.msg_ids[0])
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.close()
     
     def test_get_execute_result(self):
         """test getting execute results from the Hub."""
         c = clientmod.Client(profile='iptest')
         t = c.ids[-1]
         cell = '\n'.join([
             'import time',
             'time.sleep(0.25)',
             '5'
         ])
         ar = c[t].execute("import time; time.sleep(1)", silent=False)
         # give the monitor time to notice the message
         time.sleep(.25)
         ahr = self.client.get_result(ar.msg_ids[0])
-        print ar.get(), ahr.get(), ar._single_result, ahr._single_result
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEqual(ahr.get().pyout, ar.get().pyout)
         ar2 = self.client.get_result(ar.msg_ids[0])
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.close()
     
     def test_ids_list(self):
         """test client.ids"""
         ids = self.client.ids
         self.assertEqual(ids, self.client._ids)
         self.assertFalse(ids is self.client._ids)
         ids.remove(ids[-1])
         self.assertNotEqual(ids, self.client._ids)
     
     def test_queue_status(self):
         ids = self.client.ids
         id0 = ids[0]
         qs = self.client.queue_status(targets=id0)
         self.assertTrue(isinstance(qs, dict))
         self.assertEqual(sorted(qs.keys()), ['completed', 'queue', 'tasks'])
         allqs = self.client.queue_status()
         self.assertTrue(isinstance(allqs, dict))
         intkeys = list(allqs.keys())
         intkeys.remove('unassigned')
         self.assertEqual(sorted(intkeys), sorted(self.client.ids))
         unassigned = allqs.pop('unassigned')
         for eid,qs in allqs.items():
             self.assertTrue(isinstance(qs, dict))
             self.assertEqual(sorted(qs.keys()), ['completed', 'queue', 'tasks'])
 
     def test_shutdown(self):
         ids = self.client.ids
         id0 = ids[0]
         self.client.shutdown(id0, block=True)
         while id0 in self.client.ids:
             time.sleep(0.1)
             self.client.spin()
         
         self.assertRaises(IndexError, lambda : self.client[id0])
         
     def test_result_status(self):
         pass
         # to be written
     
     def test_db_query_dt(self):
         """test db query by date"""
         hist = self.client.hub_history()
         middle = self.client.db_query({'msg_id' : hist[len(hist)//2]})[0]
         tic = middle['submitted']
         before = self.client.db_query({'submitted' : {'$lt' : tic}})
         after = self.client.db_query({'submitted' : {'$gte' : tic}})
         self.assertEqual(len(before)+len(after),len(hist))
         for b in before:
             self.assertTrue(b['submitted'] < tic)
         for a in after:
             self.assertTrue(a['submitted'] >= tic)
         same = self.client.db_query({'submitted' : tic})
         for s in same:
             self.assertTrue(s['submitted'] == tic)
     
     def test_db_query_keys(self):
         """test extracting subset of record keys"""
         found = self.client.db_query({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])
         for rec in found:
             self.assertEqual(set(rec.keys()), set(['msg_id', 'submitted', 'completed']))
     
     def test_db_query_default_keys(self):
         """default db_query excludes buffers"""
         found = self.client.db_query({'msg_id': {'$ne' : ''}})
         for rec in found:
             keys = set(rec.keys())
             self.assertFalse('buffers' in keys, "'buffers' should not be in: %s" % keys)
             self.assertFalse('result_buffers' in keys, "'result_buffers' should not be in: %s" % keys)
     
     def test_db_query_msg_id(self):
         """ensure msg_id is always in db queries"""
         found = self.client.db_query({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
         found = self.client.db_query({'msg_id': {'$ne' : ''}},keys=['submitted'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
         found = self.client.db_query({'msg_id': {'$ne' : ''}},keys=['msg_id'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
     
     def test_db_query_get_result(self):
         """pop in db_query shouldn't pop from result itself"""
         self.client[:].apply_sync(lambda : 1)
         found = self.client.db_query({'msg_id': {'$ne' : ''}})
         rc2 = clientmod.Client(profile='iptest')
         # If this bug is not fixed, this call will hang:
         ar = rc2.get_result(self.client.history[-1])
         ar.wait(2)
         self.assertTrue(ar.ready())
         ar.get()
         rc2.close()
     
     def test_db_query_in(self):
         """test db query with '$in','$nin' operators"""
         hist = self.client.hub_history()
         even = hist[::2]
         odd = hist[1::2]
         recs = self.client.db_query({ 'msg_id' : {'$in' : even}})
         found = [ r['msg_id'] for r in recs ]
         self.assertEqual(set(even), set(found))
         recs = self.client.db_query({ 'msg_id' : {'$nin' : even}})
         found = [ r['msg_id'] for r in recs ]
         self.assertEqual(set(odd), set(found))
     
     def test_hub_history(self):
         hist = self.client.hub_history()
         recs = self.client.db_query({ 'msg_id' : {"$ne":''}})
         recdict = {}
         for rec in recs:
             recdict[rec['msg_id']] = rec
         
         latest = datetime(1984,1,1)
         for msg_id in hist:
             rec = recdict[msg_id]
             newt = rec['submitted']
             self.assertTrue(newt >= latest)
             latest = newt
         ar = self.client[-1].apply_async(lambda : 1)
         ar.get()
         time.sleep(0.25)
         self.assertEqual(self.client.hub_history()[-1:],ar.msg_ids)
     
     def _wait_for_idle(self):
         """wait for an engine to become idle, according to the Hub"""
         rc = self.client
         
         # step 1. wait for all requests to be noticed
         # timeout 5s, polling every 100ms
         msg_ids = set(rc.history)
         hub_hist = rc.hub_history()
         for i in range(50):
             if msg_ids.difference(hub_hist):
                 time.sleep(0.1)
                 hub_hist = rc.hub_history()
             else:
                 break
         
         self.assertEqual(len(msg_ids.difference(hub_hist)), 0)
         
         # step 2. wait for all requests to be done
         # timeout 5s, polling every 100ms
         qs = rc.queue_status()
         for i in range(50):
             if qs['unassigned'] or any(qs[eid]['tasks'] for eid in rc.ids):
                 time.sleep(0.1)
                 qs = rc.queue_status()
             else:
                 break
         
         # ensure Hub up to date:
         self.assertEqual(qs['unassigned'], 0)
         for eid in rc.ids:
             self.assertEqual(qs[eid]['tasks'], 0)
     
     
     def test_resubmit(self):
         def f():
             import random
             return random.random()
         v = self.client.load_balanced_view()
         ar = v.apply_async(f)
         r1 = ar.get(1)
         # give the Hub a chance to notice:
         self._wait_for_idle()
         ahr = self.client.resubmit(ar.msg_ids)
         r2 = ahr.get(1)
         self.assertFalse(r1 == r2)
 
     def test_resubmit_chain(self):
         """resubmit resubmitted tasks"""
         v = self.client.load_balanced_view()
         ar = v.apply_async(lambda x: x, 'x'*1024)
         ar.get()
         self._wait_for_idle()
         ars = [ar]
         
         for i in range(10):
             ar = ars[-1]
             ar2 = self.client.resubmit(ar.msg_ids)
         
         [ ar.get() for ar in ars ]
 
     def test_resubmit_header(self):
         """resubmit shouldn't clobber the whole header"""
         def f():
             import random
             return random.random()
         v = self.client.load_balanced_view()
         v.retries = 1
         ar = v.apply_async(f)
         r1 = ar.get(1)
         # give the Hub a chance to notice:
         self._wait_for_idle()
         ahr = self.client.resubmit(ar.msg_ids)
         ahr.get(1)
         time.sleep(0.5)
         records = self.client.db_query({'msg_id': {'$in': ar.msg_ids + ahr.msg_ids}}, keys='header')
         h1,h2 = [ r['header'] for r in records ]
         for key in set(h1.keys()).union(set(h2.keys())):
             if key in ('msg_id', 'date'):
                 self.assertNotEqual(h1[key], h2[key])
             else:
                 self.assertEqual(h1[key], h2[key])
 
     def test_resubmit_aborted(self):
         def f():
             import random
             return random.random()
         v = self.client.load_balanced_view()
         # restrict to one engine, so we can put a sleep
         # ahead of the task, so it will get aborted
         eid = self.client.ids[-1]
         v.targets = [eid]
         sleep = v.apply_async(time.sleep, 0.5)
         ar = v.apply_async(f)
         ar.abort()
         self.assertRaises(error.TaskAborted, ar.get)
         # Give the Hub a chance to get up to date:
         self._wait_for_idle()
         ahr = self.client.resubmit(ar.msg_ids)
         r2 = ahr.get(1)
 
     def test_resubmit_inflight(self):
         """resubmit of inflight task"""
         v = self.client.load_balanced_view()
         ar = v.apply_async(time.sleep,1)
         # give the message a chance to arrive
         time.sleep(0.2)
         ahr = self.client.resubmit(ar.msg_ids)
         ar.get(2)
         ahr.get(2)
 
     def test_resubmit_badkey(self):
         """ensure KeyError on resubmit of nonexistant task"""
         self.assertRaisesRemote(KeyError, self.client.resubmit, ['invalid'])
 
     def test_purge_hub_results(self):
         # ensure there are some tasks
         for i in range(5):
             self.client[:].apply_sync(lambda : 1)
         # Wait for the Hub to realise the result is done:
         # This prevents a race condition, where we
         # might purge a result the Hub still thinks is pending.
         self._wait_for_idle()
         rc2 = clientmod.Client(profile='iptest')
         hist = self.client.hub_history()
         ahr = rc2.get_result([hist[-1]])
         ahr.wait(10)
         self.client.purge_hub_results(hist[-1])
         newhist = self.client.hub_history()
         self.assertEqual(len(newhist)+1,len(hist))
         rc2.spin()
         rc2.close()
 
     def test_purge_local_results(self):
         # ensure there are some tasks
         res = []
         for i in range(5):
             res.append(self.client[:].apply_async(lambda : 1))
         self._wait_for_idle()
         self.client.wait(10) # wait for the results to come back
         before = len(self.client.results)
         self.assertEqual(len(self.client.metadata),before)
         self.client.purge_local_results(res[-1])
         self.assertEqual(len(self.client.results),before-len(res[-1]), msg="Not removed from results")
         self.assertEqual(len(self.client.metadata),before-len(res[-1]), msg="Not removed from metadata")
         
     def test_purge_all_hub_results(self):
         self.client.purge_hub_results('all')
         hist = self.client.hub_history()
         self.assertEqual(len(hist), 0)
 
     def test_purge_all_local_results(self):
         self.client.purge_local_results('all')
         self.assertEqual(len(self.client.results), 0, msg="Results not empty")
         self.assertEqual(len(self.client.metadata), 0, msg="metadata not empty")
 
     def test_purge_all_results(self):
         # ensure there are some tasks
         for i in range(5):
             self.client[:].apply_sync(lambda : 1)
         self.client.wait(10)
         self._wait_for_idle()
         self.client.purge_results('all')
         self.assertEqual(len(self.client.results), 0, msg="Results not empty")
         self.assertEqual(len(self.client.metadata), 0, msg="metadata not empty")
         hist = self.client.hub_history()
         self.assertEqual(len(hist), 0, msg="hub history not empty")
         
     def test_purge_everything(self):
         # ensure there are some tasks
         for i in range(5):
             self.client[:].apply_sync(lambda : 1)
         self.client.wait(10)
         self._wait_for_idle()
         self.client.purge_everything()
         # The client results
         self.assertEqual(len(self.client.results), 0, msg="Results not empty")
         self.assertEqual(len(self.client.metadata), 0, msg="metadata not empty")
         # The client "bookkeeping"
         self.assertEqual(len(self.client.session.digest_history), 0, msg="session digest not empty")
         self.assertEqual(len(self.client.history), 0, msg="client history not empty")
         # the hub results
         hist = self.client.hub_history()
         self.assertEqual(len(hist), 0, msg="hub history not empty")
         
     
     def test_spin_thread(self):
         self.client.spin_thread(0.01)
         ar = self.client[-1].apply_async(lambda : 1)
         time.sleep(0.1)
         self.assertTrue(ar.wall_time < 0.1,
             "spin should have kept wall_time < 0.1, but got %f" % ar.wall_time
         )
     
     def test_stop_spin_thread(self):
         self.client.spin_thread(0.01)
         self.client.stop_spin_thread()
         ar = self.client[-1].apply_async(lambda : 1)
         time.sleep(0.15)
         self.assertTrue(ar.wall_time > 0.1,
             "Shouldn't be spinning, but got wall_time=%f" % ar.wall_time
         )
     
     def test_activate(self):
         ip = get_ipython()
         magics = ip.magics_manager.magics
         self.assertTrue('px' in magics['line'])
         self.assertTrue('px' in magics['cell'])
         v0 = self.client.activate(-1, '0')
         self.assertTrue('px0' in magics['line'])
         self.assertTrue('px0' in magics['cell'])
         self.assertEqual(v0.targets, self.client.ids[-1])
         v0 = self.client.activate('all', 'all')
         self.assertTrue('pxall' in magics['line'])
         self.assertTrue('pxall' in magics['cell'])
         self.assertEqual(v0.targets, 'all')