Merge PR #413

Adds retries and resubmit logic to IPython.parallel.

closes gh-413

IPython/parallel/tests/test_lbview.py

@@ -0,0 +1,120
+"""test LoadBalancedView objects"""
+# -*- coding: utf-8 -*-
+#-------------------------------------------------------------------------------
+#  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 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)"""
+        # 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_abort(self):
+        view = self.view
+        ar = self.client[:].apply_async(time.sleep, .5)
+        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)

IPython/parallel/tests/test_mongodb.py

@@ -0,0 +1,37
+"""Tests for mongodb backend"""
+
+#-------------------------------------------------------------------------------
+#  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 nose import SkipTest
+
+from pymongo import Connection
+from IPython.parallel.controller.mongodb import MongoDB
+
+from . import test_db
+
+try:
+    c = Connection()
+except Exception:
+    c=None
+
+class TestMongoBackend(test_db.TestDictBackend):
+    """MongoDB backend tests"""
+
+    def create_db(self):
+        try:
+            return MongoDB(database='iptestdb', _connection=c)
+        except Exception:
+            raise SkipTest("Couldn't connect to mongodb")
+
+def teardown(self):
+    if c is not None:
+        c.drop_database('iptestdb')

docs/source/parallel/parallel_db.txt

@@ -0,0 +1,114
+.. _parallel_db:
+
+=======================
+IPython's Task Database
+=======================
+
+The IPython Hub stores all task requests and results in a database. Currently supported backends
+are: MongoDB, SQLite (the default), and an in-memory DictDB.  The most common use case for
+this is clients requesting results for tasks they did not submit, via:
+
+.. sourcecode:: ipython
+
+    In [1]: rc.get_result(task_id)
+
+However, since we have this DB backend, we provide a direct query method in the :class:`client`
+for users who want deeper introspection into their task history. The :meth:`db_query` method of
+the Client is modeled after MongoDB queries, so if you have used MongoDB it should look
+familiar.  In fact, when the MongoDB backend is in use, the query is relayed directly.  However,
+when using other backends, the interface is emulated and only a subset of queries is possible.
+
+.. seealso::
+
+    MongoDB query docs: http://www.mongodb.org/display/DOCS/Querying
+
+:meth:`Client.db_query` takes a dictionary query object, with keys from the TaskRecord key list,
+and values of either exact values to test, or MongoDB queries, which are dicts of The form:
+``{'operator' : 'argument(s)'}``. There is also an optional `keys` argument, that specifies
+which subset of keys should be retrieved. The default is to retrieve all keys excluding the
+request and result buffers. :meth:`db_query` returns a list of TaskRecord dicts. Also like
+MongoDB, the `msg_id` key will always be included, whether requested or not.
+
+TaskRecord keys:
+
+=============== =============== =============
+Key             Type            Description
+=============== =============== =============
+msg_id          uuid(bytes)     The msg ID
+header          dict            The request header
+content         dict            The request content (likely empty)
+buffers         list(bytes)     buffers containing serialized request objects
+submitted       datetime        timestamp for time of submission (set by client)
+client_uuid     uuid(bytes)     IDENT of client's socket
+engine_uuid     uuid(bytes)     IDENT of engine's socket
+started         datetime        time task began execution on engine
+completed       datetime        time task finished execution (success or failure) on engine
+resubmitted     datetime        time of resubmission (if applicable)
+result_header   dict            header for result
+result_content  dict            content for result
+result_buffers  list(bytes)     buffers containing serialized request objects
+queue           bytes           The name of the queue for the task ('mux' or 'task')
+pyin            <unused>        Python input (unused)
+pyout           <unused>        Python output (unused)
+pyerr           <unused>        Python traceback (unused)
+stdout          str             Stream of stdout data
+stderr          str             Stream of stderr data
+
+=============== =============== =============
+
+MongoDB operators we emulate on all backends:
+
+==========  =================
+Operator    Python equivalent
+==========  =================
+  '$in'       in
+  '$nin'      not in
+  '$eq'       ==
+  '$ne'       !=
+  '$ge'       >
+  '$gte'      >=
+  '$le'       <
+  '$lte'      <=
+==========  =================
+
+
+The DB Query is useful for two primary cases:
+
+1. deep polling of task status or metadata
+2. selecting a subset of tasks, on which to perform a later operation (e.g. wait on result, purge records, resubmit,...)
+
+Example Queries
+===============
+
+
+To get all msg_ids that are not completed, only retrieving their ID and start time:
+
+.. sourcecode:: ipython
+
+    In [1]: incomplete = rc.db_query({'complete' : None}, keys=['msg_id', 'started'])
+
+All jobs started in the last hour by me:
+
+.. sourcecode:: ipython
+
+    In [1]: from datetime import datetime, timedelta
+
+    In [2]: hourago = datetime.now() - timedelta(1./24)
+
+    In [3]: recent = rc.db_query({'started' : {'$gte' : hourago },
+                                    'client_uuid' : rc.session.session})
+
+All jobs started more than an hour ago, by clients *other than me*:
+
+.. sourcecode:: ipython
+
+    In [3]: recent = rc.db_query({'started' : {'$le' : hourago },
+                                    'client_uuid' : {'$ne' : rc.session.session}})
+
+Result headers for all jobs on engine 3 or 4:
+
+.. sourcecode:: ipython
+
+    In [1]: uuids = map(rc._engines.get, (3,4))
+
+    In [2]: hist34 = rc.db_query({'engine_uuid' : {'$in' : uuids }, keys='result_header')

IPython/parallel/client/client.py

@@ -1041,6 +1041,68 class Client(HasTraits):
             ar.wait()
         
         return ar
+
+    @spin_first
+    def resubmit(self, indices_or_msg_ids=None, subheader=None, block=None):
+        """Resubmit one or more tasks.
+
+        in-flight tasks may not be resubmitted.
+
+        Parameters
+        ----------
+
+        indices_or_msg_ids : integer history index, str msg_id, or list of either
+            The indices or msg_ids of indices to be retrieved
+
+        block : bool
+            Whether to wait for the result to be done
+
+        Returns
+        -------
+
+        AsyncHubResult
+            A subclass of AsyncResult that retrieves results from the Hub
+
+        """
+        block = self.block if block is None else block
+        if indices_or_msg_ids is None:
+            indices_or_msg_ids = -1
+
+        if not isinstance(indices_or_msg_ids, (list,tuple)):
+            indices_or_msg_ids = [indices_or_msg_ids]
+
+        theids = []
+        for id in indices_or_msg_ids:
+            if isinstance(id, int):
+                id = self.history[id]
+            if not isinstance(id, str):
+                raise TypeError("indices must be str or int, not %r"%id)
+            theids.append(id)
+
+        for msg_id in theids:
+            self.outstanding.discard(msg_id)
+            if msg_id in self.history:
+                self.history.remove(msg_id)
+            self.results.pop(msg_id, None)
+            self.metadata.pop(msg_id, None)
+        content = dict(msg_ids = theids)
+
+        self.session.send(self._query_socket, 'resubmit_request', content)
+
+        zmq.select([self._query_socket], [], [])
+        idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK)
+        if self.debug:
+            pprint(msg)
+        content = msg['content']
+        if content['status'] != 'ok':
+            raise self._unwrap_exception(content)
+
+        ar = AsyncHubResult(self, msg_ids=theids)
+
+        if block:
+            ar.wait()
+
+        return ar
     
     @spin_first
     def result_status(self, msg_ids, status_only=True):
@@ -1255,9 +1317,11 class Client(HasTraits):
         query : mongodb query dict
             The search dict. See mongodb query docs for details.
         keys : list of strs [optional]
-            THe subset of keys to be returned.  The default is to fetch everything.
+            The subset of keys to be returned.  The default is to fetch everything but buffers.
             'msg_id' will *always* be included.
         """
+        if isinstance(keys, basestring):
+            keys = [keys]
         content = dict(query=query, keys=keys)
         self.session.send(self._query_socket, "db_request", content=content)
         idents, msg = self.session.recv(self._query_socket, 0)

IPython/parallel/client/view.py

@@ -19,7 +19,7 from types import ModuleType
 import zmq
 
 from IPython.testing import decorators as testdec
-from IPython.utils.traitlets import HasTraits, Any, Bool, List, Dict, Set, Int, Instance, CFloat
+from IPython.utils.traitlets import HasTraits, Any, Bool, List, Dict, Set, Int, Instance, CFloat, CInt
 
 from IPython.external.decorator import decorator
 
@@ -791,9 +791,10 class LoadBalancedView(View):
     follow=Any()
     after=Any()
     timeout=CFloat()
+    retries = CInt(0)
     
     _task_scheme = Any()
-    _flag_names = List(['targets', 'block', 'track', 'follow', 'after', 'timeout'])
+    _flag_names = List(['targets', 'block', 'track', 'follow', 'after', 'timeout', 'retries'])
     
     def __init__(self, client=None, socket=None, **flags):
         super(LoadBalancedView, self).__init__(client=client, socket=socket, **flags)
@@ -851,7 +852,7 class LoadBalancedView(View):
             whether to create a MessageTracker to allow the user to 
             safely edit after arrays and buffers during non-copying
             sends.
-        #
+
         after : Dependency or collection of msg_ids
             Only for load-balanced execution (targets=None)
             Specify a list of msg_ids as a time-based dependency.
@@ -869,6 +870,9 class LoadBalancedView(View):
             Specify an amount of time (in seconds) for the scheduler to
             wait for dependencies to be met before failing with a
             DependencyTimeout.
+
+        retries : int
+            Number of times a task will be retried on failure.
         """
         
         super(LoadBalancedView, self).set_flags(**kwargs)
@@ -892,7 +896,7 class LoadBalancedView(View):
     @save_ids
     def _really_apply(self, f, args=None, kwargs=None, block=None, track=None,
                                         after=None, follow=None, timeout=None,
-                                        targets=None):
+                                        targets=None, retries=None):
         """calls f(*args, **kwargs) on a remote engine, returning the result.
         
         This method temporarily sets all of `apply`'s flags for a single call.
@@ -933,10 +937,11 class LoadBalancedView(View):
             raise RuntimeError(msg)
         
         if self._task_scheme == 'pure':
-            # pure zmq scheme doesn't support dependencies
-            msg = "Pure ZMQ scheduler doesn't support dependencies"
-            if (follow or after):
-                # hard fail on DAG dependencies
+            # pure zmq scheme doesn't support extra features
+            msg = "Pure ZMQ scheduler doesn't support the following flags:"
+            "follow, after, retries, targets, timeout"
+            if (follow or after or retries or targets or timeout):
+                # hard fail on Scheduler flags
                 raise RuntimeError(msg)
             if isinstance(f, dependent):
                 # soft warn on functional dependencies
@@ -948,10 +953,14 class LoadBalancedView(View):
         block = self.block if block is None else block
         track = self.track if track is None else track
         after = self.after if after is None else after
+        retries = self.retries if retries is None else retries
         follow = self.follow if follow is None else follow
         timeout = self.timeout if timeout is None else timeout
         targets = self.targets if targets is None else targets
         
+        if not isinstance(retries, int):
+            raise TypeError('retries must be int, not %r'%type(retries))
+
         if targets is None:
             idents = []
         else:
@@ -959,7 +968,7 class LoadBalancedView(View):
         
         after = self._render_dependency(after)
         follow = self._render_dependency(follow)
-        subheader = dict(after=after, follow=follow, timeout=timeout, targets=idents)
+        subheader = dict(after=after, follow=follow, timeout=timeout, targets=idents, retries=retries)
         
         msg = self.client.send_apply_message(self._socket, f, args, kwargs, track=track,
                                 subheader=subheader)

IPython/parallel/controller/dictdb.py

@@ -146,7 +146,7 class DictDB(BaseDB):
         """Remove a record from the DB."""
         matches = self._match(check)
         for m in matches:
-            del self._records[m]
+            del self._records[m['msg_id']]
         
     def drop_record(self, msg_id):
         """Remove a record from the DB."""

IPython/parallel/controller/hub.py

@@ -268,8 +268,15 class HubFactory(RegistrationFactory):
             }
         self.log.debug("Hub engine addrs: %s"%self.engine_info)
         self.log.debug("Hub client addrs: %s"%self.client_info)
+
+        # resubmit stream
+        r = ZMQStream(ctx.socket(zmq.XREQ), loop)
+        url = util.disambiguate_url(self.client_info['task'][-1])
+        r.setsockopt(zmq.IDENTITY, self.session.session)
+        r.connect(url)
+
         self.hub = Hub(loop=loop, session=self.session, monitor=sub, heartmonitor=self.heartmonitor,
-                query=q, notifier=n, db=self.db,
+                query=q, notifier=n, resubmit=r, db=self.db,
                 engine_info=self.engine_info, client_info=self.client_info,
                 logname=self.log.name)
     
@@ -315,8 +322,9 class Hub(LoggingFactory):
     loop=Instance(ioloop.IOLoop)
     query=Instance(ZMQStream)
     monitor=Instance(ZMQStream)
-    heartmonitor=Instance(HeartMonitor)
     notifier=Instance(ZMQStream)
+    resubmit=Instance(ZMQStream)
+    heartmonitor=Instance(HeartMonitor)
     db=Instance(object)
     client_info=Dict()
     engine_info=Dict()
@@ -379,6 +387,9 class Hub(LoggingFactory):
                                 'connection_request': self.connection_request,
         }
         
+        # ignore resubmit replies
+        self.resubmit.on_recv(lambda msg: None, copy=False)
+
         self.log.info("hub::created hub")
     
     @property
@@ -452,31 +463,31 class Hub(LoggingFactory):
     def dispatch_monitor_traffic(self, msg):
         """all ME and Task queue messages come through here, as well as
         IOPub traffic."""
-        self.log.debug("monitor traffic: %s"%msg[:2])
+        self.log.debug("monitor traffic: %r"%msg[:2])
         switch = msg[0]
         idents, msg = self.session.feed_identities(msg[1:])
         if not idents:
-            self.log.error("Bad Monitor Message: %s"%msg)
+            self.log.error("Bad Monitor Message: %r"%msg)
             return
         handler = self.monitor_handlers.get(switch, None)
         if handler is not None:
             handler(idents, msg)
         else:
-            self.log.error("Invalid monitor topic: %s"%switch)
+            self.log.error("Invalid monitor topic: %r"%switch)
         
     
     def dispatch_query(self, msg):
         """Route registration requests and queries from clients."""
         idents, msg = self.session.feed_identities(msg)
         if not idents:
-            self.log.error("Bad Query Message: %s"%msg)
+            self.log.error("Bad Query Message: %r"%msg)
             return
         client_id = idents[0]
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
             content = error.wrap_exception()
-            self.log.error("Bad Query Message: %s"%msg, exc_info=True)
+            self.log.error("Bad Query Message: %r"%msg, exc_info=True)
             self.session.send(self.query, "hub_error", ident=client_id, 
                     content=content)
             return
@@ -484,16 +495,17 class Hub(LoggingFactory):
         # print client_id, header, parent, content
         #switch on message type:
         msg_type = msg['msg_type']
-        self.log.info("client::client %s requested %s"%(client_id, msg_type))
+        self.log.info("client::client %r requested %r"%(client_id, msg_type))
         handler = self.query_handlers.get(msg_type, None)
         try:
-            assert handler is not None, "Bad Message Type: %s"%msg_type
+            assert handler is not None, "Bad Message Type: %r"%msg_type
         except:
             content = error.wrap_exception()
-            self.log.error("Bad Message Type: %s"%msg_type, exc_info=True)
+            self.log.error("Bad Message Type: %r"%msg_type, exc_info=True)
             self.session.send(self.query, "hub_error", ident=client_id, 
                     content=content)
             return
+
         else:
             handler(idents, msg)
             
@@ -560,9 +572,9 class Hub(LoggingFactory):
             # it's posible iopub arrived first:
             existing = self.db.get_record(msg_id)
             for key,evalue in existing.iteritems():
-                rvalue = record[key]
+                rvalue = record.get(key, None)
                 if evalue and rvalue and evalue != rvalue:
-                    self.log.error("conflicting initial state for record: %s:%s <> %s"%(msg_id, rvalue, evalue))
+                    self.log.warn("conflicting initial state for record: %r:%r <%r> %r"%(msg_id, rvalue, key, evalue))
                 elif evalue and not rvalue:
                     record[key] = evalue
             self.db.update_record(msg_id, record)
@@ -648,10 +660,22 class Hub(LoggingFactory):
         try:
             # it's posible iopub arrived first:
             existing = self.db.get_record(msg_id)
+            if existing['resubmitted']:
+                for key in ('submitted', 'client_uuid', 'buffers'):
+                    # don't clobber these keys on resubmit
+                    # submitted and client_uuid should be different
+                    # and buffers might be big, and shouldn't have changed
+                    record.pop(key)
+                    # still check content,header which should not change
+                    # but are not expensive to compare as buffers
+
             for key,evalue in existing.iteritems():
-                rvalue = record[key]
+                if key.endswith('buffers'):
+                    # don't compare buffers
+                    continue
+                rvalue = record.get(key, None)
                 if evalue and rvalue and evalue != rvalue:
-                    self.log.error("conflicting initial state for record: %s:%s <> %s"%(msg_id, rvalue, evalue))
+                    self.log.warn("conflicting initial state for record: %r:%r <%r> %r"%(msg_id, rvalue, key, evalue))
                 elif evalue and not rvalue:
                     record[key] = evalue
             self.db.update_record(msg_id, record)
@@ -1042,42 +1066,99 class Hub(LoggingFactory):
             except Exception:
                 reply = error.wrap_exception()
         else:
-            for msg_id in msg_ids:
-                if msg_id in self.all_completed:
-                    self.db.drop_record(msg_id)
-                else:
-                    if msg_id in self.pending:
-                        try:
-                            raise IndexError("msg pending: %r"%msg_id)
-                        except:
-                            reply = error.wrap_exception()
-                    else:
+            pending = filter(lambda m: m in self.pending, msg_ids)
+            if pending:
+                try:
+                    raise IndexError("msg pending: %r"%pending[0])
+                except:
+                    reply = error.wrap_exception()
+            else:
+                try:
+                    self.db.drop_matching_records(dict(msg_id={'$in':msg_ids}))
+                except Exception:
+                    reply = error.wrap_exception()
+
+            if reply['status'] == 'ok':
+                eids = content.get('engine_ids', [])
+                for eid in eids:
+                    if eid not in self.engines:
                         try:
-                            raise IndexError("No such msg: %r"%msg_id)
+                            raise IndexError("No such engine: %i"%eid)
                         except:
                             reply = error.wrap_exception()
-                    break
-            eids = content.get('engine_ids', [])
-            for eid in eids:
-                if eid not in self.engines:
+                        break
+                    msg_ids = self.completed.pop(eid)
+                    uid = self.engines[eid].queue
                     try:
-                        raise IndexError("No such engine: %i"%eid)
-                    except:
+                        self.db.drop_matching_records(dict(engine_uuid=uid, completed={'$ne':None}))
+                    except Exception:
                         reply = error.wrap_exception()
-                    break
-                msg_ids = self.completed.pop(eid)
-                uid = self.engines[eid].queue
-                try:
-                    self.db.drop_matching_records(dict(engine_uuid=uid, completed={'$ne':None}))
-                except Exception:
-                    reply = error.wrap_exception()
-                    break
+                        break
         
         self.session.send(self.query, 'purge_reply', content=reply, ident=client_id)
     
-    def resubmit_task(self, client_id, msg, buffers):
-        """Resubmit a task."""
-        raise NotImplementedError
+    def resubmit_task(self, client_id, msg):
+        """Resubmit one or more tasks."""
+        def finish(reply):
+            self.session.send(self.query, 'resubmit_reply', content=reply, ident=client_id)
+
+        content = msg['content']
+        msg_ids = content['msg_ids']
+        reply = dict(status='ok')
+        try:
+            records = self.db.find_records({'msg_id' : {'$in' : msg_ids}}, keys=[
+                'header', 'content', 'buffers'])
+        except Exception:
+            self.log.error('db::db error finding tasks to resubmit', exc_info=True)
+            return finish(error.wrap_exception())
+
+        # validate msg_ids
+        found_ids = [ rec['msg_id'] for rec in records ]
+        invalid_ids = filter(lambda m: m in self.pending, found_ids)
+        if len(records) > len(msg_ids):
+            try:
+                raise RuntimeError("DB appears to be in an inconsistent state."
+                    "More matching records were found than should exist")
+            except Exception:
+                return finish(error.wrap_exception())
+        elif len(records) < len(msg_ids):
+            missing = [ m for m in msg_ids if m not in found_ids ]
+            try:
+                raise KeyError("No such msg(s): %s"%missing)
+            except KeyError:
+                return finish(error.wrap_exception())
+        elif invalid_ids:
+            msg_id = invalid_ids[0]
+            try:
+                raise ValueError("Task %r appears to be inflight"%(msg_id))
+            except Exception:
+                return finish(error.wrap_exception())
+
+        # clear the existing records
+        rec = empty_record()
+        map(rec.pop, ['msg_id', 'header', 'content', 'buffers', 'submitted'])
+        rec['resubmitted'] = datetime.now()
+        rec['queue'] = 'task'
+        rec['client_uuid'] = client_id[0]
+        try:
+            for msg_id in msg_ids:
+                self.all_completed.discard(msg_id)
+                self.db.update_record(msg_id, rec)
+        except Exception:
+            self.log.error('db::db error upating record', exc_info=True)
+            reply = error.wrap_exception()
+        else:
+            # send the messages
+            for rec in records:
+                header = rec['header']
+                msg = self.session.msg(header['msg_type'])
+                msg['content'] = rec['content']
+                msg['header'] = header
+                msg['msg_id'] = rec['msg_id']
+                self.session.send(self.resubmit, msg, buffers=rec['buffers'])
+
+        finish(dict(status='ok'))
+
     
     def _extract_record(self, rec):
         """decompose a TaskRecord dict into subsection of reply for get_result"""
@@ -1124,12 +1205,20 class Hub(LoggingFactory):
         for msg_id in msg_ids:
             if msg_id in self.pending:
                 pending.append(msg_id)
-            elif msg_id in self.all_completed or msg_id in records:
+            elif msg_id in self.all_completed:
                 completed.append(msg_id)
                 if not statusonly:
                     c,bufs = self._extract_record(records[msg_id])
                     content[msg_id] = c
                     buffers.extend(bufs)
+            elif msg_id in records:
+                if rec['completed']:
+                    completed.append(msg_id)
+                    c,bufs = self._extract_record(records[msg_id])
+                    content[msg_id] = c
+                    buffers.extend(bufs)
+                else:
+                    pending.append(msg_id)
             else:
                 try:
                     raise KeyError('No such message: '+msg_id)

IPython/parallel/controller/mongodb.py

@@ -6,12 +6,10
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
-from datetime import datetime
-
 from pymongo import Connection
 from pymongo.binary import Binary
 
-from IPython.utils.traitlets import Dict, List, CUnicode
+from IPython.utils.traitlets import Dict, List, CUnicode, CStr, Instance
 
 from .dictdb import BaseDB
 
@@ -25,15 +23,20 class MongoDB(BaseDB):
     connection_args = List(config=True) # args passed to pymongo.Connection
     connection_kwargs = Dict(config=True) # kwargs passed to pymongo.Connection
     database = CUnicode(config=True) # name of the mongodb database
-    _table = Dict()
+
+    _connection = Instance(Connection) # pymongo connection
     
     def __init__(self, **kwargs):
         super(MongoDB, self).__init__(**kwargs)
-        self._connection = Connection(*self.connection_args, **self.connection_kwargs)
+        if self._connection is None:
+            self._connection = Connection(*self.connection_args, **self.connection_kwargs)
         if not self.database:
             self.database = self.session
         self._db = self._connection[self.database]
         self._records = self._db['task_records']
+        self._records.ensure_index('msg_id', unique=True)
+        self._records.ensure_index('submitted') # for sorting history
+        # for rec in self._records.find
     
     def _binary_buffers(self, rec):
         for key in ('buffers', 'result_buffers'):
@@ -45,18 +48,21 class MongoDB(BaseDB):
         """Add a new Task Record, by msg_id."""
         # print rec
         rec = self._binary_buffers(rec)
-        obj_id = self._records.insert(rec)
-        self._table[msg_id] = obj_id
+        self._records.insert(rec)
     
     def get_record(self, msg_id):
         """Get a specific Task Record, by msg_id."""
-        return self._records.find_one(self._table[msg_id])
+        r = self._records.find_one({'msg_id': msg_id})
+        if not r:
+            # r will be '' if nothing is found
+            raise KeyError(msg_id)
+        return r
     
     def update_record(self, msg_id, rec):
         """Update the data in an existing record."""
         rec = self._binary_buffers(rec)
-        obj_id = self._table[msg_id]
-        self._records.update({'_id':obj_id}, {'$set': rec})
+
+        self._records.update({'msg_id':msg_id}, {'$set': rec})
     
     def drop_matching_records(self, check):
         """Remove a record from the DB."""
@@ -64,8 +70,7 class MongoDB(BaseDB):
         
     def drop_record(self, msg_id):
         """Remove a record from the DB."""
-        obj_id = self._table.pop(msg_id)
-        self._records.remove(obj_id)
+        self._records.remove({'msg_id':msg_id})
     
     def find_records(self, check, keys=None):
         """Find records matching a query dict, optionally extracting subset of keys.

IPython/parallel/controller/scheduler.py

@@ -137,6 +137,7 class TaskScheduler(SessionFactory):
     
     # internals:
     graph = Dict() # dict by msg_id of [ msg_ids that depend on key ]
+    retries = Dict() # dict by msg_id of retries remaining (non-neg ints)
     # waiting = List() # list of msg_ids ready to run, but haven't due to HWM
     depending = Dict() # dict by msg_id of (msg_id, raw_msg, after, follow)
     pending = Dict() # dict by engine_uuid of submitted tasks
@@ -205,6 +206,8 class TaskScheduler(SessionFactory):
         self.pending[uid] = {}
         if len(self.targets) == 1:
             self.resume_receiving()
+        # rescan the graph:
+        self.update_graph(None)
 
     def _unregister_engine(self, uid):
         """Existing engine with ident `uid` became unavailable."""
@@ -215,11 +218,11 class TaskScheduler(SessionFactory):
         # handle any potentially finished tasks:
         self.engine_stream.flush()
         
-        self.completed.pop(uid)
-        self.failed.pop(uid)
-        # don't pop destinations, because it might be used later
+        # don't pop destinations, because they might be used later
         # map(self.destinations.pop, self.completed.pop(uid))
         # map(self.destinations.pop, self.failed.pop(uid))
+
+        # prevent this engine from receiving work
         idx = self.targets.index(uid)
         self.targets.pop(idx)
         self.loads.pop(idx)
@@ -229,28 +232,40 class TaskScheduler(SessionFactory):
         if self.pending[uid]:
             dc = ioloop.DelayedCallback(lambda : self.handle_stranded_tasks(uid), 5000, self.loop)
             dc.start()
+        else:
+            self.completed.pop(uid)
+            self.failed.pop(uid)
+
     
     @logged
     def handle_stranded_tasks(self, engine):
         """Deal with jobs resident in an engine that died."""
-        lost = self.pending.pop(engine)
-        
-        for msg_id, (raw_msg, targets, MET, follow, timeout) in lost.iteritems():
-            self.all_failed.add(msg_id)
-            self.all_done.add(msg_id)
+        lost = self.pending[engine]
+        for msg_id in lost.keys():
+            if msg_id not in self.pending[engine]:
+                # prevent double-handling of messages
+                continue
+
+            raw_msg = lost[msg_id][0]
+
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.unpack_message(msg, copy=False, content=False)
             parent = msg['header']
-            idents = [idents[0],engine]+idents[1:]
-            # print (idents)
+            idents = [engine, idents[0]]
+
+            # build fake error reply
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(engine, msg_id))
             except:
                 content = error.wrap_exception()
-            msg = self.session.send(self.client_stream, 'apply_reply', content, 
-                                                    parent=parent, ident=idents)
-            self.session.send(self.mon_stream, msg, ident=['outtask']+idents)
-            self.update_graph(msg_id)
+            msg = self.session.msg('apply_reply', content, parent=parent, subheader={'status':'error'})
+            raw_reply = map(zmq.Message, self.session.serialize(msg, ident=idents))
+            # and dispatch it
+            self.dispatch_result(raw_reply)
+
+        # finally scrub completed/failed lists
+        self.completed.pop(engine)
+        self.failed.pop(engine)
     
     
     #-----------------------------------------------------------------------
@@ -277,6 +292,8 class TaskScheduler(SessionFactory):
         
         # targets
         targets = set(header.get('targets', []))
+        retries = header.get('retries', 0)
+        self.retries[msg_id] = retries
         
         # time dependencies
         after = Dependency(header.get('after', []))
@@ -315,7 +332,9 class TaskScheduler(SessionFactory):
             # time deps already met, try to run
             if not self.maybe_run(msg_id, *args):
                 # can't run yet
-                self.save_unmet(msg_id, *args)
+                if msg_id not in self.all_failed:
+                    # could have failed as unreachable
+                    self.save_unmet(msg_id, *args)
         else:
             self.save_unmet(msg_id, *args)
     
@@ -328,7 +347,7 class TaskScheduler(SessionFactory):
             if msg_id in self.depending:
                 raw,after,targets,follow,timeout = self.depending[msg_id]
                 if timeout and timeout < now:
-                    self.fail_unreachable(msg_id, timeout=True)
+                    self.fail_unreachable(msg_id, error.TaskTimeout)
                 
     @logged
     def fail_unreachable(self, msg_id, why=error.ImpossibleDependency):
@@ -369,7 +388,7 class TaskScheduler(SessionFactory):
             # we need a can_run filter
             def can_run(idx):
                 # check hwm
-                if self.loads[idx] == self.hwm:
+                if self.hwm and self.loads[idx] == self.hwm:
                     return False
                 target = self.targets[idx]
                 # check blacklist
@@ -382,6 +401,7 class TaskScheduler(SessionFactory):
                 return follow.check(self.completed[target], self.failed[target])
             
             indices = filter(can_run, range(len(self.targets)))
+
             if not indices:
                 # couldn't run
                 if follow.all:
@@ -395,12 +415,14 class TaskScheduler(SessionFactory):
                     for m in follow.intersection(relevant):
                         dests.add(self.destinations[m])
                     if len(dests) > 1:
+                        self.depending[msg_id] = (raw_msg, targets, after, follow, timeout)
                         self.fail_unreachable(msg_id)
                         return False
                 if targets:
                     # check blacklist+targets for impossibility
                     targets.difference_update(blacklist)
                     if not targets or not targets.intersection(self.targets):
+                        self.depending[msg_id] = (raw_msg, targets, after, follow, timeout)
                         self.fail_unreachable(msg_id)
                         return False
                 return False
@@ -454,20 +476,34 class TaskScheduler(SessionFactory):
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.unpack_message(msg, content=False, copy=False)
             engine = idents[0]
-            idx = self.targets.index(engine)
-            self.finish_job(idx)
+            try:
+                idx = self.targets.index(engine)
+            except ValueError:
+                pass # skip load-update for dead engines
+            else:
+                self.finish_job(idx)
         except Exception:
             self.log.error("task::Invaid result: %s"%raw_msg, exc_info=True)
             return
 
         header = msg['header']
+        parent = msg['parent_header']
         if header.get('dependencies_met', True):
             success = (header['status'] == 'ok')
-            self.handle_result(idents, msg['parent_header'], raw_msg, success)
-            # send to Hub monitor
-            self.mon_stream.send_multipart(['outtask']+raw_msg, copy=False)
+            msg_id = parent['msg_id']
+            retries = self.retries[msg_id]
+            if not success and retries > 0:
+                # failed
+                self.retries[msg_id] = retries - 1
+                self.handle_unmet_dependency(idents, parent)
+            else:
+                del self.retries[msg_id]
+                # relay to client and update graph
+                self.handle_result(idents, parent, raw_msg, success)
+                # send to Hub monitor
+                self.mon_stream.send_multipart(['outtask']+raw_msg, copy=False)
         else:
-            self.handle_unmet_dependency(idents, msg['parent_header'])
+            self.handle_unmet_dependency(idents, parent)
         
     @logged
     def handle_result(self, idents, parent, raw_msg, success=True):
@@ -511,13 +547,19 class TaskScheduler(SessionFactory):
             self.depending[msg_id] = args
             self.fail_unreachable(msg_id)
         elif not self.maybe_run(msg_id, *args):
-            # resubmit failed, put it back in our dependency tree
-            self.save_unmet(msg_id, *args)
+            # resubmit failed
+            if msg_id not in self.all_failed:
+                # put it back in our dependency tree
+                self.save_unmet(msg_id, *args)
         
         if self.hwm:
-            idx = self.targets.index(engine)
-            if self.loads[idx] == self.hwm-1:
-                self.update_graph(None)
+            try:
+                idx = self.targets.index(engine)
+            except ValueError:
+                pass # skip load-update for dead engines
+            else:
+                if self.loads[idx] == self.hwm-1:
+                    self.update_graph(None)
         
         
     
@@ -526,7 +568,7 class TaskScheduler(SessionFactory):
         """dep_id just finished. Update our dependency
         graph and submit any jobs that just became runable.
         
-        Called with dep_id=None to update graph for hwm, but without finishing
+        Called with dep_id=None to update entire graph for hwm, but without finishing
         a task.
         """
         # print ("\n\n***********")
@@ -538,9 +580,11 class TaskScheduler(SessionFactory):
         # print ("\n\n***********\n\n")
         # update any jobs that depended on the dependency
         jobs = self.graph.pop(dep_id, [])
-        # if we have HWM and an engine just become no longer full
-        # recheck *all* jobs:
-        if self.hwm and any( [ load==self.hwm-1 for load in self.loads]):
+
+        # recheck *all* jobs if
+        # a) we have HWM and an engine just become no longer full
+        # or b) dep_id was given as None
+        if dep_id is None or self.hwm and any( [ load==self.hwm-1 for load in self.loads ]):
             jobs = self.depending.keys()
         
         for msg_id in jobs:

IPython/parallel/controller/sqlitedb.py

@@ -27,16 +27,20 operators = {
  '$lt' : "<",
  '$gt' : ">",
  # null is handled weird with ==,!=
- '$eq' : "IS",
- '$ne' : "IS NOT",
+ '$eq' : "=",
+ '$ne' : "!=",
  '$lte': "<=",
  '$gte': ">=",
- '$in' : ('IS', ' OR '),
- '$nin': ('IS NOT', ' AND '),
+ '$in' : ('=', ' OR '),
+ '$nin': ('!=', ' AND '),
  # '$all': None,
  # '$mod': None,
  # '$exists' : None
 }
+null_operators = {
+'=' : "IS NULL",
+'!=' : "IS NOT NULL",
+}
 
 def _adapt_datetime(dt):
     return dt.strftime(ISO8601)
@@ -205,17 +209,27 class SQLiteDB(BaseDB):
                         raise KeyError("Unsupported operator: %r"%test)
                     if isinstance(op, tuple):
                         op, join = op
-                    expr = "%s %s ?"%(name, op)
-                    if isinstance(value, (tuple,list)):
-                        expr = '( %s )'%( join.join([expr]*len(value)) )
-                        args.extend(value)
+                    
+                    if value is None and op in null_operators:
+                            expr = "%s %s"%null_operators[op]
                     else:
-                        args.append(value)
+                        expr = "%s %s ?"%(name, op)
+                        if isinstance(value, (tuple,list)):
+                            if op in null_operators and any([v is None for v in value]):
+                                # equality tests don't work with NULL
+                                raise ValueError("Cannot use %r test with NULL values on SQLite backend"%test)
+                            expr = '( %s )'%( join.join([expr]*len(value)) )
+                            args.extend(value)
+                        else:
+                            args.append(value)
                     expressions.append(expr)
             else:
                 # it's an equality check
-                expressions.append("%s IS ?"%name)
-                args.append(sub_check)
+                if sub_check is None:
+                    expressions.append("%s IS NULL")
+                else:
+                    expressions.append("%s = ?"%name)
+                    args.append(sub_check)
         
         expr = " AND ".join(expressions)
         return expr, args

IPython/parallel/streamsession.py

@@ -176,6 +176,37 class StreamSession(object):
         header = extract_header(msg_or_header)
         return header.get('key', None) == self.key
             
+
+    def serialize(self, msg, ident=None):
+        content = msg.get('content', {})
+        if content is None:
+            content = self.none
+        elif isinstance(content, dict):
+            content = self.pack(content)
+        elif isinstance(content, bytes):
+            # content is already packed, as in a relayed message
+            pass
+        elif isinstance(content, unicode):
+            # should be bytes, but JSON often spits out unicode
+            content = content.encode('utf8')
+        else:
+            raise TypeError("Content incorrect type: %s"%type(content))
+
+        to_send = []
+
+        if isinstance(ident, list):
+            # accept list of idents
+            to_send.extend(ident)
+        elif ident is not None:
+            to_send.append(ident)
+        to_send.append(DELIM)
+        if self.key is not None:
+            to_send.append(self.key)
+        to_send.append(self.pack(msg['header']))
+        to_send.append(self.pack(msg['parent_header']))
+        to_send.append(content)
+
+        return to_send
         
     def send(self, stream, msg_or_type, content=None, buffers=None, parent=None, subheader=None, ident=None, track=False):
         """Build and send a message via stream or socket.
@@ -221,33 +252,11 class StreamSession(object):
             # we got a Message, not a msg_type
             # don't build a new Message
             msg = msg_or_type
-            content = msg['content']
         else:
             msg = self.msg(msg_or_type, content, parent, subheader)
         
         buffers = [] if buffers is None else buffers
-        to_send = []
-        if isinstance(ident, list):
-            # accept list of idents
-            to_send.extend(ident)
-        elif ident is not None:
-            to_send.append(ident)
-        to_send.append(DELIM)
-        if self.key is not None:
-            to_send.append(self.key)
-        to_send.append(self.pack(msg['header']))
-        to_send.append(self.pack(msg['parent_header']))
-        
-        if content is None:
-            content = self.none
-        elif isinstance(content, dict):
-            content = self.pack(content)
-        elif isinstance(content, bytes):
-            # content is already packed, as in a relayed message
-            pass
-        else:
-            raise TypeError("Content incorrect type: %s"%type(content))
-        to_send.append(content)
+        to_send = self.serialize(msg, ident)
         flag = 0
         if buffers:
             flag = zmq.SNDMORE

IPython/parallel/tests/__init__.py

@@ -48,7 +48,7 class TestProcessLauncher(LocalProcessLauncher):
 def setup():
     cp = TestProcessLauncher()
     cp.cmd_and_args = ipcontroller_cmd_argv + \
-                ['--profile', 'iptest', '--log-level', '99', '-r', '--usethreads']
+                ['--profile', 'iptest', '--log-level', '99', '-r']
     cp.start()
     launchers.append(cp)
     cluster_dir = os.path.join(get_ipython_dir(), 'cluster_iptest')

IPython/parallel/tests/test_client.py

@@ -212,3 +212,33 class TestClient(ClusterTestCase):
         time.sleep(0.25)
         self.assertEquals(self.client.hub_history()[-1:],ar.msg_ids)
     
+    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)
+        ahr = self.client.resubmit(ar.msg_ids)
+        r2 = ahr.get(1)
+        self.assertFalse(r1 == r2)
+
+    def test_resubmit_inflight(self):
+        """ensure ValueError on 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)
+        self.assertRaisesRemote(ValueError, self.client.resubmit, ar.msg_ids)
+        ar.get(2)
+
+    def test_resubmit_badkey(self):
+        """ensure KeyError on resubmit of nonexistant task"""
+        self.assertRaisesRemote(KeyError, self.client.resubmit, ['invalid'])
+
+    def test_purge_results(self):
+        hist = self.client.hub_history()
+        self.client.purge_results(hist)
+        newhist = self.client.hub_history()
+        self.assertTrue(len(newhist) == 0)
+

IPython/parallel/tests/test_db.py

@@ -15,10 +15,7
 import tempfile
 import time
 
-import uuid
-
 from datetime import datetime, timedelta
-from random import choice, randint
 from unittest import TestCase
 
 from nose import SkipTest
@@ -157,6 +154,13 class TestDictBackend(TestCase):
         self.db.update_record(msg_id, dict(completed=datetime.now()))
         rec = self.db.get_record(msg_id)
         self.assertTrue(isinstance(rec['completed'], datetime))
+
+    def test_drop_matching(self):
+        msg_ids = self.load_records(10)
+        query = {'msg_id' : {'$in':msg_ids}}
+        self.db.drop_matching_records(query)
+        recs = self.db.find_records(query)
+        self.assertTrue(len(recs)==0)
             
 class TestSQLiteBackend(TestDictBackend):
     def create_db(self):
@@ -164,19 +168,3 class TestSQLiteBackend(TestDictBackend):
     
     def tearDown(self):
         self.db._db.close()
-
-# optional MongoDB test
-try:
-    from IPython.parallel.controller.mongodb import MongoDB
-except ImportError:
-    pass
-else:
-    class TestMongoBackend(TestDictBackend):
-        def create_db(self):
-            try:
-                return MongoDB(database='iptestdb')
-            except Exception:
-                raise SkipTest("Couldn't connect to mongodb instance")
-            
-        def tearDown(self):
-            self.db._connection.drop_database('iptestdb')

IPython/parallel/tests/test_view.py

@@ -21,7 +21,7 import zmq
 from IPython import parallel  as pmod
 from IPython.parallel import error
 from IPython.parallel import AsyncResult, AsyncHubResult, AsyncMapResult
-from IPython.parallel import LoadBalancedView, DirectView
+from IPython.parallel import DirectView
 from IPython.parallel.util import interactive
 
 from IPython.parallel.tests import add_engines
@@ -33,18 +33,6 def setup():
 
 class TestView(ClusterTestCase):
     
-    def test_z_crash_task(self):
-        """test graceful handling of engine death (balanced)"""
-        # self.add_engines(1)
-        ar = self.client[-1].apply_async(crash)
-        self.assertRaisesRemote(error.EngineError, ar.get)
-        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_z_crash_mux(self):
         """test graceful handling of engine death (direct)"""
         # self.add_engines(1)

IPython/testing/iptest.py

@@ -199,6 +199,7 def make_exclude():
     
     if not have['pymongo']:
         exclusions.append(ipjoin('parallel', 'controller', 'mongodb'))
+        exclusions.append(ipjoin('parallel', 'tests', 'test_mongodb'))
 
     if not have['matplotlib']:
         exclusions.extend([ipjoin('lib', 'pylabtools'),

docs/source/parallel/index.txt

@@ -12,6 +12,7 Using IPython for parallel computing
    parallel_multiengine.txt
    parallel_task.txt
    parallel_mpi.txt
+   parallel_db.txt
    parallel_security.txt
    parallel_winhpc.txt
    parallel_demos.txt

docs/source/parallel/parallel_task.txt

@@ -292,6 +292,7 you can skip using Dependency objects, and just pass msg_ids or AsyncResult obje
 
 
 
+
 Impossible Dependencies
 ***********************
 
@@ -313,6 +314,27 The basic cases that are checked:
     This analysis has not been proven to be rigorous, so it is likely possible for tasks
     to become impossible to run in obscure situations, so a timeout may be a good choice.
 
+
+Retries and Resubmit
+====================
+
+Retries
+-------
+
+Another flag for tasks is `retries`.  This is an integer, specifying how many times
+a task should be resubmitted after failure.  This is useful for tasks that should still run
+if their engine was shutdown, or may have some statistical chance of failing.  The default
+is to not retry tasks.
+
+Resubmit
+--------
+
+Sometimes you may want to re-run a task. This could be because it failed for some reason, and
+you have fixed the error, or because you want to restore the cluster to an interrupted state.
+For this, the :class:`Client` has a :meth:`rc.resubmit` method.  This simply takes one or more
+msg_ids, and returns an :class:`AsyncHubResult` for the result(s).  You cannot resubmit
+a task that is pending - only those that have finished, either successful or unsuccessful.
+
 .. _parallel_schedulers:
 
 Schedulers
@@ -391,6 +413,8 Disabled features when using the ZMQ Scheduler:
     TODO: performance comparisons
 
 
+
+
 More details
 ============