AsyncResult.__getattr__ shouldn't raise TimeoutError

This causes problems for things that use hasattr, e.g. list(ar) checking
for `__length_hint__`.

tests added for getattr/getitem behavior

IPython/parallel/client/asyncresult.py

@@ -1,395 +1,396
 """AsyncResult objects for the client
 
 Authors:
 
 * MinRK
 """
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 
 import time
 
 from zmq import MessageTracker
 
 from IPython.external.decorator import decorator
 from IPython.parallel import error
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 # global empty tracker that's always done:
 finished_tracker = MessageTracker()
 
 @decorator
 def check_ready(f, self, *args, **kwargs):
     """Call spin() to sync state prior to calling the method."""
     self.wait(0)
     if not self._ready:
         raise error.TimeoutError("result not ready")
     return f(self, *args, **kwargs)
 
 class AsyncResult(object):
     """Class for representing results of non-blocking calls.
 
     Provides the same interface as :py:class:`multiprocessing.pool.AsyncResult`.
     """
 
     msg_ids = None
     _targets = None
     _tracker = None
     _single_result = False
 
     def __init__(self, client, msg_ids, fname='unknown', targets=None, tracker=None):
         if isinstance(msg_ids, basestring):
             # always a list
             msg_ids = [msg_ids]
         if tracker is None:
             # default to always done
             tracker = finished_tracker
         self._client = client
         self.msg_ids = msg_ids
         self._fname=fname
         self._targets = targets
         self._tracker = tracker
         self._ready = False
         self._success = None
+        self._metadata = None
         if len(msg_ids) == 1:
             self._single_result = not isinstance(targets, (list, tuple))
         else:
             self._single_result = False
 
     def __repr__(self):
         if self._ready:
             return "<%s: finished>"%(self.__class__.__name__)
         else:
             return "<%s: %s>"%(self.__class__.__name__,self._fname)
 
 
     def _reconstruct_result(self, res):
         """Reconstruct our result from actual result list (always a list)
 
         Override me in subclasses for turning a list of results
         into the expected form.
         """
         if self._single_result:
             return res[0]
         else:
             return res
 
     def get(self, timeout=-1):
         """Return the result when it arrives.
 
         If `timeout` is not ``None`` and the result does not arrive within
         `timeout` seconds then ``TimeoutError`` is raised. If the
         remote call raised an exception then that exception will be reraised
         by get() inside a `RemoteError`.
         """
         if not self.ready():
             self.wait(timeout)
 
         if self._ready:
             if self._success:
                 return self._result
             else:
                 raise self._exception
         else:
             raise error.TimeoutError("Result not ready.")
 
     def ready(self):
         """Return whether the call has completed."""
         if not self._ready:
             self.wait(0)
         return self._ready
 
     def wait(self, timeout=-1):
         """Wait until the result is available or until `timeout` seconds pass.
 
         This method always returns None.
         """
         if self._ready:
             return
         self._ready = self._client.wait(self.msg_ids, timeout)
         if self._ready:
             try:
                 results = map(self._client.results.get, self.msg_ids)
                 self._result = results
                 if self._single_result:
                     r = results[0]
                     if isinstance(r, Exception):
                         raise r
                 else:
                     results = error.collect_exceptions(results, self._fname)
                 self._result = self._reconstruct_result(results)
             except Exception, e:
                 self._exception = e
                 self._success = False
             else:
                 self._success = True
             finally:
                 self._metadata = map(self._client.metadata.get, self.msg_ids)
 
 
     def successful(self):
         """Return whether the call completed without raising an exception.
 
         Will raise ``AssertionError`` if the result is not ready.
         """
         assert self.ready()
         return self._success
 
     #----------------------------------------------------------------
     # Extra methods not in mp.pool.AsyncResult
     #----------------------------------------------------------------
 
     def get_dict(self, timeout=-1):
         """Get the results as a dict, keyed by engine_id.
 
         timeout behavior is described in `get()`.
         """
 
         results = self.get(timeout)
         engine_ids = [ md['engine_id'] for md in self._metadata ]
         bycount = sorted(engine_ids, key=lambda k: engine_ids.count(k))
         maxcount = bycount.count(bycount[-1])
         if maxcount > 1:
             raise ValueError("Cannot build dict, %i jobs ran on engine #%i"%(
                     maxcount, bycount[-1]))
 
         return dict(zip(engine_ids,results))
 
     @property
     def result(self):
         """result property wrapper for `get(timeout=0)`."""
         return self.get()
 
     # abbreviated alias:
     r = result
 
     @property
     @check_ready
     def metadata(self):
         """property for accessing execution metadata."""
         if self._single_result:
             return self._metadata[0]
         else:
             return self._metadata
 
     @property
     def result_dict(self):
         """result property as a dict."""
         return self.get_dict()
 
     def __dict__(self):
         return self.get_dict(0)
 
     def abort(self):
         """abort my tasks."""
         assert not self.ready(), "Can't abort, I am already done!"
         return self.client.abort(self.msg_ids, targets=self._targets, block=True)
 
     @property
     def sent(self):
         """check whether my messages have been sent."""
         return self._tracker.done
 
     def wait_for_send(self, timeout=-1):
         """wait for pyzmq send to complete.
 
         This is necessary when sending arrays that you intend to edit in-place.
         `timeout` is in seconds, and will raise TimeoutError if it is reached
         before the send completes.
         """
         return self._tracker.wait(timeout)
 
     #-------------------------------------
     # dict-access
     #-------------------------------------
 
     @check_ready
     def __getitem__(self, key):
         """getitem returns result value(s) if keyed by int/slice, or metadata if key is str.
         """
         if isinstance(key, int):
             return error.collect_exceptions([self._result[key]], self._fname)[0]
         elif isinstance(key, slice):
             return error.collect_exceptions(self._result[key], self._fname)
         elif isinstance(key, basestring):
             values = [ md[key] for md in self._metadata ]
             if self._single_result:
                 return values[0]
             else:
                 return values
         else:
             raise TypeError("Invalid key type %r, must be 'int','slice', or 'str'"%type(key))
 
-    @check_ready
     def __getattr__(self, key):
         """getattr maps to getitem for convenient attr access to metadata."""
-        if key not in self._metadata[0].keys():
+        try:
+            return self.__getitem__(key)
+        except (error.TimeoutError, KeyError):
             raise AttributeError("%r object has no attribute %r"%(
                     self.__class__.__name__, key))
-        return self.__getitem__(key)
 
     # asynchronous iterator:
     def __iter__(self):
         if self._single_result:
             raise TypeError("AsyncResults with a single result are not iterable.")
         try:
             rlist = self.get(0)
         except error.TimeoutError:
             # wait for each result individually
             for msg_id in self.msg_ids:
                 ar = AsyncResult(self._client, msg_id, self._fname)
                 yield ar.get()
         else:
             # already done
             for r in rlist:
                 yield r
 
 
 
 class AsyncMapResult(AsyncResult):
     """Class for representing results of non-blocking gathers.
 
     This will properly reconstruct the gather.
     
     This class is iterable at any time, and will wait on results as they come.
     
     If ordered=False, then the first results to arrive will come first, otherwise
     results will be yielded in the order they were submitted.
     
     """
 
     def __init__(self, client, msg_ids, mapObject, fname='', ordered=True):
         AsyncResult.__init__(self, client, msg_ids, fname=fname)
         self._mapObject = mapObject
         self._single_result = False
         self.ordered = ordered
 
     def _reconstruct_result(self, res):
         """Perform the gather on the actual results."""
         return self._mapObject.joinPartitions(res)
 
     # asynchronous iterator:
     def __iter__(self):
         it = self._ordered_iter if self.ordered else self._unordered_iter
         for r in it():
             yield r
 
     # asynchronous ordered iterator:
     def _ordered_iter(self):
         """iterator for results *as they arrive*, preserving submission order."""
         try:
             rlist = self.get(0)
         except error.TimeoutError:
             # wait for each result individually
             for msg_id in self.msg_ids:
                 ar = AsyncResult(self._client, msg_id, self._fname)
                 rlist = ar.get()
                 try:
                     for r in rlist:
                         yield r
                 except TypeError:
                     # flattened, not a list
                     # this could get broken by flattened data that returns iterables
                     # but most calls to map do not expose the `flatten` argument
                     yield rlist
         else:
             # already done
             for r in rlist:
                 yield r
 
     # asynchronous unordered iterator:
     def _unordered_iter(self):
         """iterator for results *as they arrive*, on FCFS basis, ignoring submission order."""
         try:
             rlist = self.get(0)
         except error.TimeoutError:
             pending = set(self.msg_ids)
             while pending:
                 try:
                     self._client.wait(pending, 1e-3)
                 except error.TimeoutError:
                     # ignore timeout error, because that only means
                     # *some* jobs are outstanding
                     pass
                 # update ready set with those no longer outstanding:
                 ready = pending.difference(self._client.outstanding)
                 # update pending to exclude those that are finished
                 pending = pending.difference(ready)
                 while ready:
                     msg_id = ready.pop()
                     ar = AsyncResult(self._client, msg_id, self._fname)
                     rlist = ar.get()
                     try:
                         for r in rlist:
                             yield r
                     except TypeError:
                         # flattened, not a list
                         # this could get broken by flattened data that returns iterables
                         # but most calls to map do not expose the `flatten` argument
                         yield rlist
         else:
             # already done
             for r in rlist:
                 yield r
 
 
 
 class AsyncHubResult(AsyncResult):
     """Class to wrap pending results that must be requested from the Hub.
 
     Note that waiting/polling on these objects requires polling the Hubover the network,
     so use `AsyncHubResult.wait()` sparingly.
     """
 
     def wait(self, timeout=-1):
         """wait for result to complete."""
         start = time.time()
         if self._ready:
             return
         local_ids = filter(lambda msg_id: msg_id in self._client.outstanding, self.msg_ids)
         local_ready = self._client.wait(local_ids, timeout)
         if local_ready:
             remote_ids = filter(lambda msg_id: msg_id not in self._client.results, self.msg_ids)
             if not remote_ids:
                 self._ready = True
             else:
                 rdict = self._client.result_status(remote_ids, status_only=False)
                 pending = rdict['pending']
                 while pending and (timeout < 0 or time.time() < start+timeout):
                     rdict = self._client.result_status(remote_ids, status_only=False)
                     pending = rdict['pending']
                     if pending:
                         time.sleep(0.1)
                 if not pending:
                     self._ready = True
         if self._ready:
             try:
                 results = map(self._client.results.get, self.msg_ids)
                 self._result = results
                 if self._single_result:
                     r = results[0]
                     if isinstance(r, Exception):
                         raise r
                 else:
                     results = error.collect_exceptions(results, self._fname)
                 self._result = self._reconstruct_result(results)
             except Exception, e:
                 self._exception = e
                 self._success = False
             else:
                 self._success = True
             finally:
                 self._metadata = map(self._client.metadata.get, self.msg_ids)
 
 __all__ = ['AsyncResult', 'AsyncMapResult', 'AsyncHubResult']
  No newline at end of file

IPython/parallel/tests/test_asyncresult.py

@@ -1,73 +1,115
 """Tests for asyncresult.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 IPython.parallel.error import TimeoutError
 
 from IPython.parallel.tests import add_engines
 from .clienttest import ClusterTestCase
 
 def setup():
     add_engines(2)
 
 def wait(n):
     import time
     time.sleep(n)
     return n
 
 class AsyncResultTest(ClusterTestCase):
 
     def test_single_result(self):
         eid = self.client.ids[-1]
         ar = self.client[eid].apply_async(lambda : 42)
         self.assertEquals(ar.get(), 42)
         ar = self.client[[eid]].apply_async(lambda : 42)
         self.assertEquals(ar.get(), [42])
         ar = self.client[-1:].apply_async(lambda : 42)
         self.assertEquals(ar.get(), [42])
 
     def test_get_after_done(self):
         ar = self.client[-1].apply_async(lambda : 42)
         ar.wait()
         self.assertTrue(ar.ready())
         self.assertEquals(ar.get(), 42)
         self.assertEquals(ar.get(), 42)
 
     def test_get_before_done(self):
         ar = self.client[-1].apply_async(wait, 0.1)
         self.assertRaises(TimeoutError, ar.get, 0)
         ar.wait(0)
         self.assertFalse(ar.ready())
         self.assertEquals(ar.get(), 0.1)
 
     def test_get_after_error(self):
         ar = self.client[-1].apply_async(lambda : 1/0)
         ar.wait(10)
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
         self.assertRaisesRemote(ZeroDivisionError, ar.get_dict)
     
     def test_get_dict(self):
         n = len(self.client)
         ar = self.client[:].apply_async(lambda : 5)
         self.assertEquals(ar.get(), [5]*n)
         d = ar.get_dict()
         self.assertEquals(sorted(d.keys()), sorted(self.client.ids))
         for eid,r in d.iteritems():
             self.assertEquals(r, 5)
+    
+    def test_list_amr(self):
+        ar = self.client.load_balanced_view().map_async(wait, [0.1]*5)
+        rlist = list(ar)
+    
+    def test_getattr(self):
+        ar = self.client[:].apply_async(wait, 0.5)
+        self.assertRaises(AttributeError, lambda : ar._foo)
+        self.assertRaises(AttributeError, lambda : ar.__length_hint__())
+        self.assertRaises(AttributeError, lambda : ar.foo)
+        self.assertRaises(AttributeError, lambda : ar.engine_id)
+        self.assertFalse(hasattr(ar, '__length_hint__'))
+        self.assertFalse(hasattr(ar, 'foo'))
+        self.assertFalse(hasattr(ar, 'engine_id'))
+        ar.get(5)
+        self.assertRaises(AttributeError, lambda : ar._foo)
+        self.assertRaises(AttributeError, lambda : ar.__length_hint__())
+        self.assertRaises(AttributeError, lambda : ar.foo)
+        self.assertTrue(isinstance(ar.engine_id, list))
+        self.assertEquals(ar.engine_id, ar['engine_id'])
+        self.assertFalse(hasattr(ar, '__length_hint__'))
+        self.assertFalse(hasattr(ar, 'foo'))
+        self.assertTrue(hasattr(ar, 'engine_id'))
+
+    def test_getitem(self):
+        ar = self.client[:].apply_async(wait, 0.5)
+        self.assertRaises(TimeoutError, lambda : ar['foo'])
+        self.assertRaises(TimeoutError, lambda : ar['engine_id'])
+        ar.get(5)
+        self.assertRaises(KeyError, lambda : ar['foo'])
+        self.assertTrue(isinstance(ar['engine_id'], list))
+        self.assertEquals(ar.engine_id, ar['engine_id'])
+    
+    def test_single_result(self):
+        ar = self.client[-1].apply_async(wait, 0.5)
+        self.assertRaises(TimeoutError, lambda : ar['foo'])
+        self.assertRaises(TimeoutError, lambda : ar['engine_id'])
+        self.assertTrue(ar.get(5) == 0.5)
+        self.assertTrue(isinstance(ar['engine_id'], int))
+        self.assertTrue(isinstance(ar.engine_id, int))
+        self.assertEquals(ar.engine_id, ar['engine_id'])
+
 

IPython/parallel/tests/test_lbview.py

@@ -1,165 +1,167
 # -*- coding: utf-8 -*-
 """test LoadBalancedView 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 time
 
 import zmq
 from nose import SkipTest
 
 from IPython import parallel  as pmod
 from IPython.parallel import error
 
 from IPython.parallel.tests import add_engines
 
 from .clienttest import ClusterTestCase, crash, wait, skip_without
 
 def setup():
     add_engines(3)
 
 class TestLoadBalancedView(ClusterTestCase):
 
     def setUp(self):
         ClusterTestCase.setUp(self)
         self.view = self.client.load_balanced_view()
 
     def test_z_crash_task(self):
         """test graceful handling of engine death (balanced)"""
         raise SkipTest("crash tests disabled, due to undesirable crash reports")
         # self.add_engines(1)
         ar = self.view.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_map(self):
         def f(x):
             return x**2
         data = range(16)
         r = self.view.map_sync(f, data)
         self.assertEquals(r, map(f, data))
 
     def test_map_unordered(self):
         def f(x):
             return x**2
         def slow_f(x):
             import time
             time.sleep(0.05*x)
             return x**2
         data = range(16,0,-1)
         reference = map(f, data)
         
-        amr = self.view.map_async(f, data, ordered=False)
+        amr = self.view.map_async(slow_f, data, ordered=False)
         self.assertTrue(isinstance(amr, pmod.AsyncMapResult))
-        # check individual elements, retrieved as they come (uses __iter__)
-        astheycame = list(amr)
+        # check individual elements, retrieved as they come
+        # list comprehension uses __iter__
+        astheycame = [ r for r in amr ]
         # Ensure that at least one result came out of order:
         self.assertNotEquals(astheycame, reference, "should not have preserved order")
         self.assertEquals(sorted(astheycame, reverse=True), reference, "result corrupted")
 
     def test_map_ordered(self):
         def f(x):
             return x**2
         def slow_f(x):
             import time
             time.sleep(0.05*x)
             return x**2
         data = range(16,0,-1)
         reference = map(f, data)
         
-        amr = self.view.map_async(f, data)
+        amr = self.view.map_async(slow_f, data)
         self.assertTrue(isinstance(amr, pmod.AsyncMapResult))
-        # check individual elements, retrieved as they come (uses __iter__)
+        # check individual elements, retrieved as they come
+        # list(amr) uses __iter__
         astheycame = list(amr)
         # Ensure that results came in order
         self.assertEquals(astheycame, reference)
         self.assertEquals(amr.result, reference)
     
     def test_abort(self):
         view = self.view
         ar = self.client[:].apply_async(time.sleep, .5)
         ar = self.client[:].apply_async(time.sleep, .5)
         time.sleep(0.2)
         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_retries(self):
         add_engines(3)
         view = self.view
         view.timeout = 1 # prevent hang if this doesn't behave
         def fail():
             assert False
         for r in range(len(self.client)-1):
             with view.temp_flags(retries=r):
                 self.assertRaisesRemote(AssertionError, view.apply_sync, fail)
 
         with view.temp_flags(retries=len(self.client), timeout=0.25):
             self.assertRaisesRemote(error.TaskTimeout, view.apply_sync, fail)
 
     def test_invalid_dependency(self):
         view = self.view
         with view.temp_flags(after='12345'):
             self.assertRaisesRemote(error.InvalidDependency, view.apply_sync, lambda : 1)
 
     def test_impossible_dependency(self):
         if len(self.client) < 2:
             add_engines(2)
         view = self.client.load_balanced_view()
         ar1 = view.apply_async(lambda : 1)
         ar1.get()
         e1 = ar1.engine_id
         e2 = e1
         while e2 == e1:
             ar2 = view.apply_async(lambda : 1)
             ar2.get()
             e2 = ar2.engine_id
 
         with view.temp_flags(follow=[ar1, ar2]):
             self.assertRaisesRemote(error.ImpossibleDependency, view.apply_sync, lambda : 1)
 
 
     def test_follow(self):
         ar = self.view.apply_async(lambda : 1)
         ar.get()
         ars = []
         first_id = ar.engine_id
 
         self.view.follow = ar
         for i in range(5):
             ars.append(self.view.apply_async(lambda : 1))
         self.view.wait(ars)
         for ar in ars:
             self.assertEquals(ar.engine_id, first_id)
 
     def test_after(self):
         view = self.view
         ar = view.apply_async(time.sleep, 0.5)
         with view.temp_flags(after=ar):
             ar2 = view.apply_async(lambda : 1)
 
         ar.wait()
         ar2.wait()
         self.assertTrue(ar2.started >= ar.completed, "%s not >= %s"%(ar.started, ar.completed))