Backport PR #4722: allow purging local results as long as they are not outstanding

raise RuntimeError, not assert

closes  #4721

IPython/parallel/client/client.py

@@ -1659,15 +1659,13 b' class Client(HasTraits):'
         return msg_ids        
         
     def purge_local_results(self, jobs=[], targets=[]):
-        """Clears the client caches of results and frees such memory.
-        
+        """Clears the client caches of results and their metadata.
+
         Individual results can be purged by msg_id, or the entire
         history of specific targets can be purged.
 
-        Use `purge_local_results('all')` to scrub everything from the Clients's db.
-
-        The client must have no outstanding tasks before purging the caches.
-        Raises `AssertionError` if there are still outstanding tasks.
+        Use `purge_local_results('all')` to scrub everything from the Clients's
+        results and metadata caches.
 
         After this call all `AsyncResults` are invalid and should be discarded.
 
@@ -1681,24 +1679,30 b' class Client(HasTraits):'
 
         jobs : str or list of str or AsyncResult objects
                 the msg_ids whose results should be purged.
-        targets : int/str/list of ints/strs
-                The targets, by int_id, whose entire results are to be purged.
+        targets : int/list of ints
+                The engines, by integer ID, whose entire result histories are to be purged.
+
+        Raises
+        ------
+
+        RuntimeError : if any of the tasks to be purged are still outstanding.
 
-                default : None
         """
-        assert not self.outstanding, "Can't purge a client with outstanding tasks!"
-        
         if not targets and not jobs:
             raise ValueError("Must specify at least one of `targets` and `jobs`")
-                
+
         if jobs == 'all':
+            if self.outstanding:
+                raise RuntimeError("Can't purge outstanding tasks: %s" % self.outstanding)
             self.results.clear()
             self.metadata.clear()
-            return
         else:
-            msg_ids = []
-            msg_ids.extend(self._build_msgids_from_target(targets))
-            msg_ids.extend(self._build_msgids_from_jobs(jobs))
+            msg_ids = set()
+            msg_ids.update(self._build_msgids_from_target(targets))
+            msg_ids.update(self._build_msgids_from_jobs(jobs))
+            still_outstanding = self.outstanding.intersection(msg_ids)
+            if still_outstanding:
+                raise RuntimeError("Can't purge outstanding tasks: %s" % still_outstanding)
             map(self.results.pop, msg_ids)
             map(self.metadata.pop, msg_ids)
 

IPython/parallel/tests/test_client.py

@@ -444,7 +444,32 b' class TestClient(ClusterTestCase):'
         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_local_results_outstanding(self):
+        v = self.client[-1]
+        ar = v.apply_async(lambda : 1)
+        msg_id = ar.msg_ids[0]
+        ar.get()
+        self._wait_for_idle()
+        ar2 = v.apply_async(time.sleep, 1)
+        self.assertIn(msg_id, self.client.results)
+        self.assertIn(msg_id, self.client.metadata)
+        self.client.purge_local_results(ar)
+        self.assertNotIn(msg_id, self.client.results)
+        self.assertNotIn(msg_id, self.client.metadata)
+        with self.assertRaises(RuntimeError):
+            self.client.purge_local_results(ar2)
+        ar2.get()
+        self.client.purge_local_results(ar2)
+
+    def test_purge_all_local_results_outstanding(self):
+        v = self.client[-1]
+        ar = v.apply_async(time.sleep, 1)
+        with self.assertRaises(RuntimeError):
+            self.client.purge_local_results('all')
+        ar.get()
+        self.client.purge_local_results('all')
+
     def test_purge_all_hub_results(self):
         self.client.purge_hub_results('all')
         hist = self.client.hub_history()