wait for empty queues as well as tasks

in parallel test `_wait_for_idle`

work submitted via DirectView was ignored in `_wait_for_idle`,
so if such a task is outstanding, the condition can return prematurely.

should close  #4502

IPython/parallel/tests/test_client.py

@@ -301,9 +301,13 b' class TestClient(ClusterTestCase):'
         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"""
+        """wait for the cluster to become idle, according to the everyone."""
         rc = self.client
         
+        # step 0. wait for local results
+        # this should be sufficient 99% of the time.
+        rc.wait(timeout=5)
+        
         # step 1. wait for all requests to be noticed
         # timeout 5s, polling every 100ms
         msg_ids = set(rc.history)
@@ -321,7 +325,7 b' class TestClient(ClusterTestCase):'
         # 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):
+            if qs['unassigned'] or any(qs[eid]['tasks'] + qs[eid]['queue'] for eid in rc.ids):
                 time.sleep(0.1)
                 qs = rc.queue_status()
             else:
@@ -331,6 +335,7 b' class TestClient(ClusterTestCase):'
         self.assertEqual(qs['unassigned'], 0)
         for eid in rc.ids:
             self.assertEqual(qs[eid]['tasks'], 0)
+            self.assertEqual(qs[eid]['queue'], 0)
     
     
     def test_resubmit(self):