Merge pull request #462 from minrk/cl-7-pre

7.0 prerelease

Author: minrk

docs/source/changelog.rst

@@ -5,6 +5,19 @@ Changelog
 
 Changes in IPython Parallel
 
+7.0.0
+-----
+
+**prerelease** there are some big things coming! This is a first prerelease to get some things out there for testing
+
+- **Require Python 3.6**
+- Fix compatibility issues with ipykernel 6, jupyter-client 7
+- New prototype BroadcastScheduler with vastly improved scaling in 'do-on-all' operations,
+  c/o Tom-Olav Bøyum's Master's thesis.
+- Add :meth:`.AsyncResult.stream_output` context manager for streaming output.
+  Stream output by default in parallel magics.
+- The repo has been updated to use pre-commit, black, myst, and friends and GitHub Actions for CI, but this should not affect users, only making it a bit nicer for contributors.
+
 6.3.0
 -----
 

ipyparallel/client/view.py

@@ -864,6 +864,34 @@ def activate(self, suffix=''):
 class BroadcastView(DirectView):
     is_coalescing = Bool(False)
 
+    def _init_metadata(self, s_idents):
+        """initialize request metadata"""
+        return dict(
+            targets=s_idents,
+            is_broadcast=True,
+            is_coalescing=self.is_coalescing,
+        )
+
+    def _make_async_result(self, message_future, s_idents, **kwargs):
+        original_msg_id = message_future.msg_id
+        if not self.is_coalescing:
+            futures = []
+            for ident in s_idents:
+                msg_and_target_id = f'{original_msg_id}_{ident}'
+                future = self.client.create_message_futures(
+                    msg_and_target_id, async_result=True, track=True
+                )
+                self.client.outstanding.add(msg_and_target_id)
+                self.outstanding.add(msg_and_target_id)
+                futures.append(future[0])
+            if original_msg_id in self.outstanding:
+                self.outstanding.remove(original_msg_id)
+        else:
+            self.client.outstanding.add(original_msg_id)
+            futures = message_future
+
+        return AsyncResult(self.client, futures, owner=True, **kwargs)
+
     @sync_results
     @save_ids
     def _really_apply(
@@ -883,44 +911,62 @@ def _really_apply(
 
         s_idents = [ident.decode("utf8") for ident in idents]
 
-        metadata = dict(
-            targets=s_idents, is_broadcast=True, is_coalescing=self.is_coalescing
+        metadata = self._init_metadata(s_idents)
+        message_future = self.client.send_apply_request(
+            self._socket, pf, pargs, pkwargs, track=track, metadata=metadata
+        )
+        ar = self._make_async_result(
+            message_future, s_idents, fname=getname(f), targets=targets
         )
-        if not self.is_coalescing:
-            original_future = self.client.send_apply_request(
-                self._socket, pf, pargs, pkwargs, track=track, metadata=metadata
-            )
-            original_msg_id = original_future.msg_id
 
-            for ident in s_idents:
-                msg_and_target_id = f'{original_msg_id}_{ident}'
-                future = self.client.create_message_futures(
-                    msg_and_target_id, async_result=True, track=True
-                )
-                self.client.outstanding.add(msg_and_target_id)
-                self.outstanding.add(msg_and_target_id)
-                futures.append(future[0])
-            if original_msg_id in self.outstanding:
-                self.outstanding.remove(original_msg_id)
-        else:
-            message_future = self.client.send_apply_request(
-                self._socket, pf, pargs, pkwargs, track=track, metadata=metadata
-            )
-            self.client.outstanding.add(message_future.msg_id)
-            futures = message_future
+        if block:
+            try:
+                return ar.get()
+            except KeyboardInterrupt:
+                pass
+        return ar
 
-        ar = AsyncResult(
-            self.client, futures, fname=getname(f), targets=_targets, owner=True
+    @sync_results
+    @save_ids
+    def execute(self, code, silent=True, targets=None, block=None):
+        """Executes `code` on `targets` in blocking or nonblocking manner.
+
+        ``execute`` is always `bound` (affects engine namespace)
+
+        Parameters
+        ----------
+        code : str
+            the code string to be executed
+        block : bool
+            whether or not to wait until done to return
+            default: self.block
+        """
+        block = self.block if block is None else block
+        targets = self.targets if targets is None else targets
+
+        _idents, _targets = self.client._build_targets(targets)
+        s_idents = [ident.decode("utf8") for ident in _idents]
+
+        metadata = self._init_metadata(s_idents)
+        message_future = self.client.send_execute_request(
+            self._socket,
+            code,
+            silent=silent,
+            metadata=metadata,
+        )
+        ar = self._make_async_result(
+            message_future, s_idents, fname='execute', targets=_targets
         )
         if block:
             try:
-                return ar.get()
+                ar.get()
+                ar.wait_for_output()
             except KeyboardInterrupt:
                 pass
         return ar
 
     def map(self, f, *sequences, **kwargs):
-        pass
+        raise NotImplementedError("BroadcastView.map not yet implemented")
 
 
 class LoadBalancedView(View):

ipyparallel/controller/broadcast_scheduler.py

@@ -125,12 +125,11 @@ def dispatch_result(self, raw_msg):
             msg = self.session.deserialize(msg, content=False, copy=False)
             outgoing_id = idents[0]
 
-        except:
+        except Exception:
             self.log.error(
                 f'broadcast::Invalid broadcast msg: {raw_msg}', exc_info=True
             )
             return
-
         original_msg_id = msg['metadata']['original_msg_id']
         is_coalescing = msg['metadata']['is_coalescing']
         if is_coalescing:

ipyparallel/tests/clienttest.py

@@ -167,6 +167,8 @@ def setUp(self):
         self.engines = []
 
     def tearDown(self):
+        self.client[:].use_pickle()
+
         # self.client.clear(block=True)
         # close fds:
         for e in filter(lambda e: e.poll() is not None, launchers):

ipyparallel/tests/test_joblib.py

@@ -36,8 +36,6 @@ def test_default_backend(self):
             p = Parallel(backend='ipyparallel')
             assert p._backend._view.client is self.client
 
-        self.client[:].use_pickle()
-
     def test_register_backend(self):
         view = self.client.load_balanced_view()
         view.register_joblib_backend('view')

ipyparallel/tests/test_view_broadcast.py

@@ -0,0 +1,67 @@
+"""test BroadcastView objects"""
+import pytest
+
+from . import test_view
+
+
+needs_map = pytest.mark.xfail(reason="map not yet implemented")
+
+
+@pytest.mark.usefixtures('ipython')
+class TestBroadcastView(test_view.TestView):
+    def setUp(self):
+        super().setUp()
+        self._broadcast_view_used = False
+        # use broadcast view for direct API
+        real_direct_view = self.client.direct_view
+
+        def broadcast_or_direct(targets):
+            if isinstance(targets, int):
+                return real_direct_view(targets)
+            else:
+                self._broadcast_view_used = True
+                return self.client.broadcast_view(targets)
+
+        self.client.direct_view = broadcast_or_direct
+
+    def tearDown(self):
+        super().tearDown()
+        # note that a test didn't use a broadcast view
+        if not self._broadcast_view_used:
+            pytest.skip("No broadcast view used")
+
+    @pytest.mark.xfail(reason="aborted replies missing metadata")
+    def test_abort(self):
+        pass
+
+    @pytest.mark.xfail(reason="aborted replies missing metadata")
+    def test_abort_all(self):
+        pass
+
+    @needs_map
+    def test_map(self):
+        pass
+
+    @needs_map
+    def test_map_ref(self):
+        pass
+
+    @needs_map
+    def test_map_reference(self):
+        pass
+
+    @needs_map
+    def test_map_iterable(self):
+        pass
+
+    @needs_map
+    def test_map_empty_sequence(self):
+        pass
+
+    @needs_map
+    def test_map_numpy(self):
+        pass
+
+    @pytest.mark.xfail(reason="Tracking gets disconnected from original message")
+    def test_scatter_tracked(self):
+        pass