Merge pull request #521 from minrk/public-controller-engines

make launchers public attributes

Author: minrk

ipyparallel/cluster/app.py

@@ -187,12 +187,12 @@ def start(self):
             ):
                 profile = abbreviate_profile_dir(cluster.profile_dir)
                 cluster_id = cluster.cluster_id
-                running = bool(cluster._controller)
+                running = bool(cluster.controller)
                 # TODO: URL?
                 engines = 0
-                if cluster._engine_sets:
+                if cluster.engines:
                     engines = sum(
-                        engine_set.n for engine_set in cluster._engine_sets.values()
+                        engine_set.n for engine_set in cluster.engines.values()
                     )
 
                 launcher = cluster.engine_launcher_class.__name__
@@ -351,7 +351,7 @@ async def start_engines(self):
     def watch_engines(self):
         """Watch for early engine shutdown"""
         # FIXME: public API to get launcher instances?
-        self.engine_launcher = next(iter(self.cluster._engine_sets.values()))
+        self.engine_launcher = next(iter(self.cluster.engines.values()))
 
         if not self.early_shutdown:
             self.engine_launcher.on_stop(self.engines_stopped)
@@ -477,7 +477,7 @@ async def start_cluster(self):
                 f"Leaving cluster running: {self.cluster.cluster_file}", file=sys.stderr
             )
             self.loop.add_callback(self.loop.stop)
-        self.cluster._controller.on_stop(self.controller_stopped)
+        self.cluster.controller.on_stop(self.controller_stopped)
         self.watch_engines()
 
     def controller_stopped(self, stop_data):

ipyparallel/cluster/cluster.py

@@ -54,7 +54,7 @@ def _atexit_cleanup_clusters(*args):
         if not cluster.shutdown_atexit:
             # overridden after register
             continue
-        if cluster._controller or cluster._engine_sets:
+        if cluster.controller or cluster.engines:
             print(f"Stopping cluster {cluster}", file=sys.stderr)
             cluster.stop_cluster_sync()
 
@@ -263,8 +263,8 @@ def _default_log(self):
             return traitlets.log.get_logger()
 
     # private state
-    _controller = Any()
-    _engine_sets = Dict()
+    controller = Any()
+    engines = Dict()
 
     def __init__(self, **kwargs):
         """Construct a Cluster"""
@@ -275,7 +275,7 @@ def __init__(self, **kwargs):
     def __del__(self):
         if not self.shutdown_atexit:
             return
-        if self._controller or self._engine_sets:
+        if self.controller or self.engines:
             self.stop_cluster_sync()
 
     def __repr__(self):
@@ -295,10 +295,10 @@ def __repr__(self):
                 profile_dir = "~" + profile_dir[len(home_dir) :]
             fields["profile_dir"] = repr(profile_dir)
 
-        if self._controller:
+        if self.controller:
             fields["controller"] = "<running>"
-        if self._engine_sets:
-            fields["engine_sets"] = list(self._engine_sets)
+        if self.engines:
+            fields["engine_sets"] = list(self.engines)
 
         fields_str = ', '.join(f"{key}={value}" for key, value in fields.items())
 
@@ -316,20 +316,20 @@ def _cls_str(cls):
 
         cluster_info["class"] = _cls_str(self.__class__)
 
-        if self._controller:
+        if self.controller:
             d["controller"] = {
                 "class": _cls_str(self.controller_launcher_class),
                 "state": None,
             }
-            if self._controller:
-                d["controller"]["state"] = self._controller.to_dict()
+            if self.controller:
+                d["controller"]["state"] = self.controller.to_dict()
 
         d["engines"] = {
             "class": _cls_str(self.engine_launcher_class),
             "sets": {},
         }
         sets = d["engines"]["sets"]
-        for engine_set_id, engine_launcher in self._engine_sets.items():
+        for engine_set_id, engine_launcher in self.engines.items():
             sets[engine_set_id] = engine_launcher.to_dict()
         return d
 
@@ -358,13 +358,13 @@ def from_dict(cls, d, **kwargs):
             controller_info = d["controller"]
             cls = self.controller_launcher_class = import_item(controller_info["class"])
             if controller_info["state"]:
-                self._controller = cls.from_dict(controller_info["state"], parent=self)
+                self.controller = cls.from_dict(controller_info["state"], parent=self)
 
         engine_info = d.get("engines")
         if engine_info:
             cls = self.engine_launcher_class = import_item(engine_info["class"])
             for engine_set_id, engine_state in engine_info.get("sets", {}).items():
-                self._engine_sets[engine_set_id] = cls.from_dict(
+                self.engines[engine_set_id] = cls.from_dict(
                     engine_state,
                     engine_set_id=engine_set_id,
                     parent=self,
@@ -431,7 +431,7 @@ def update_cluster_file(self):
             # setting cluster_file='' disables saving to disk
             return
 
-        if not self._controller and not self._engine_sets:
+        if not self.controller and not self.engines:
             self.remove_cluster_file()
         else:
             self.write_cluster_file()
@@ -444,17 +444,17 @@ async def start_controller(self, **kwargs):
         # start controller
         # retrieve connection info
         # webhook?
-        if self._controller is not None:
+        if self.controller is not None:
             raise RuntimeError(
-                "controller is already running. Call stop_controller() first."
+                "controller is already running. Call stopcontroller() first."
             )
 
         if self.shutdown_atexit:
             _atexit_clusters.add(self)
             if not _atexit_cleanup_clusters.registered:
                 atexit.register(_atexit_cleanup_clusters)
 
-        self._controller = controller = self.controller_launcher_class(
+        self.controller = controller = self.controller_launcher_class(
             work_dir=u'.',
             parent=self,
             log=self.log,
@@ -486,10 +486,10 @@ def add_args(args):
 
         if controller_args is not None:
             # ensure we trigger trait observers after we are done
-            self._controller.controller_args = list(controller_args)
+            self.controller.controller_args = list(controller_args)
 
-        self._controller.on_stop(self._controller_stopped)
-        r = self._controller.start()
+        self.controller.on_stop(self._controller_stopped)
+        r = self.controller.start()
         if inspect.isawaitable(r):
             await r
 
@@ -507,7 +507,7 @@ async def start_engines(self, n=None, engine_set_id=None, **kwargs):
         # TODO: send engines connection info
         if engine_set_id is None:
             engine_set_id = f"{int(time.time())}-{''.join(random.choice(_suffix_chars) for i in range(4))}"
-        engine_set = self._engine_sets[engine_set_id] = self.engine_launcher_class(
+        engine_set = self.engines[engine_set_id] = self.engine_launcher_class(
             work_dir=u'.',
             parent=self,
             log=self.log,
@@ -550,17 +550,17 @@ async def stop_engines(self, engine_set_id=None):
         all engines are stopped.
         """
         if engine_set_id is None:
-            for engine_set_id in list(self._engine_sets):
+            for engine_set_id in list(self.engines):
                 await self.stop_engines(engine_set_id)
             return
         self.log.info(f"Stopping engine(s): {engine_set_id}")
-        engine_set = self._engine_sets[engine_set_id]
+        engine_set = self.engines[engine_set_id]
         r = engine_set.stop()
         if inspect.isawaitable(r):
             await r
         # retrieve and cleanup output files
         engine_set.get_output(remove=True)
-        self._engine_sets.pop(engine_set_id)
+        self.engines.pop(engine_set_id)
         self.update_cluster_file()
 
     async def stop_engine(self, engine_id):
@@ -574,10 +574,10 @@ async def stop_engine(self, engine_id):
     async def restart_engines(self, engine_set_id=None):
         """Restart an engine set"""
         if engine_set_id is None:
-            for engine_set_id in list(self._engine_sets):
+            for engine_set_id in list(self.engines):
                 await self.restart_engines(engine_set_id)
             return
-        engine_set = self._engine_sets[engine_set_id]
+        engine_set = self.engines[engine_set_id]
         n = engine_set.n
         await self.stop_engines(engine_set_id)
         await self.start_engines(n, engine_set_id)
@@ -604,27 +604,27 @@ async def signal_engines(self, signum, engine_set_id=None):
         If no engine set is specified, signal all engine sets.
         """
         if engine_set_id is None:
-            for engine_set_id in list(self._engine_sets):
+            for engine_set_id in list(self.engines):
                 await self.signal_engines(signum, engine_set_id)
             return
         self.log.info(f"Sending signal {signum} to engine(s) {engine_set_id}")
-        engine_set = self._engine_sets[engine_set_id]
+        engine_set = self.engines[engine_set_id]
         r = engine_set.signal(signum)
         if inspect.isawaitable(r):
             await r
 
     async def stop_controller(self):
         """Stop the controller"""
-        if self._controller and self._controller.running:
+        if self.controller and self.controller.running:
             self.log.info("Stopping controller")
-            r = self._controller.stop()
+            r = self.controller.stop()
             if inspect.isawaitable(r):
                 await r
 
-        if self._controller:
-            self._controller.get_output(remove=True)
+        if self.controller:
+            self.controller.get_output(remove=True)
 
-        self._controller = None
+        self.controller = None
         self.update_cluster_file()
 
     async def stop_cluster(self):
@@ -638,7 +638,7 @@ async def connect_client(self, **client_kwargs):
         # this assumes local files exist
         from ipyparallel import Client
 
-        connection_info = self._controller.get_connection_info()
+        connection_info = self.controller.get_connection_info()
         if inspect.isawaitable(connection_info):
             connection_info = await connection_info
 

ipyparallel/tests/test_cluster.py

@@ -64,16 +64,16 @@ async def test_start_stop_controller(Cluster):
     with pytest.raises(RuntimeError):
         await cluster.start_controller()
     assert cluster.config is not None
-    assert cluster._controller.config is cluster.config
-    assert cluster._controller is not None
-    proc = cluster._controller.process
+    assert cluster.controller.config is cluster.config
+    assert cluster.controller is not None
+    proc = cluster.controller.process
     assert proc.is_running()
     with await cluster.connect_client() as rc:
         assert rc.queue_status() == {'unassigned': 0}
 
     await cluster.stop_controller()
     proc.wait(timeout=3)
-    assert cluster._controller is None
+    assert cluster.controller is None
     # stop is idempotent
     await cluster.stop_controller()
     # TODO: test file cleanup
@@ -85,16 +85,16 @@ async def test_start_stop_engines(Cluster, engine_launcher_class):
 
     n = 2
     engine_set_id = await cluster.start_engines(n)
-    assert engine_set_id in cluster._engine_sets
-    engine_set = cluster._engine_sets[engine_set_id]
+    assert engine_set_id in cluster.engines
+    engine_set = cluster.engines[engine_set_id]
     launcher_class = find_launcher_class(engine_launcher_class, "EngineSet")
     assert isinstance(engine_set, launcher_class)
 
     with await cluster.connect_client() as rc:
         rc.wait_for_engines(n, timeout=_timeout)
 
     await cluster.stop_engines(engine_set_id)
-    assert cluster._engine_sets == {}
+    assert cluster.engines == {}
     with pytest.raises(KeyError):
         await cluster.stop_engines(engine_set_id)
 
@@ -105,15 +105,15 @@ async def test_start_stop_cluster(Cluster, engine_launcher_class):
     n = 2
     cluster = Cluster(engine_launcher_class=engine_launcher_class, n=n)
     await cluster.start_cluster()
-    controller = cluster._controller
+    controller = cluster.controller
     assert controller is not None
-    assert len(cluster._engine_sets) == 1
+    assert len(cluster.engines) == 1
 
     with await cluster.connect_client() as rc:
         rc.wait_for_engines(n, timeout=_timeout)
     await cluster.stop_cluster()
-    assert cluster._controller is None
-    assert cluster._engine_sets == {}
+    assert cluster.controller is None
+    assert cluster.engines == {}
 
 
 @pytest.mark.skipif(
@@ -155,8 +155,8 @@ async def test_restart_engines(Cluster, engine_launcher_class):
     n = 2
     async with Cluster(engine_launcher_class=engine_launcher_class, n=n) as rc:
         cluster = rc.cluster
-        engine_set_id = next(iter(cluster._engine_sets))
-        engine_set = cluster._engine_sets[engine_set_id]
+        engine_set_id = next(iter(cluster.engines))
+        engine_set = cluster.engines[engine_set_id]
         assert rc.ids[:n] == list(range(n))
         before_pids = rc[:].apply_sync(os.getpid)
         await cluster.restart_engines()
@@ -240,12 +240,12 @@ async def test_to_from_dict(Cluster, engine_launcher_class):
         d = cluster.to_dict()
         cluster2 = ipp.Cluster.from_dict(d)
         assert not cluster2.shutdown_atexit
-        assert cluster2._controller is not None
-        assert cluster2._controller.process.pid == cluster._controller.process.pid
-        assert list(cluster2._engine_sets) == list(cluster._engine_sets)
+        assert cluster2.controller is not None
+        assert cluster2.controller.process.pid == cluster.controller.process.pid
+        assert list(cluster2.engines) == list(cluster.engines)
 
-        es1 = next(iter(cluster._engine_sets.values()))
-        es2 = next(iter(cluster2._engine_sets.values()))
+        es1 = next(iter(cluster.engines.values()))
+        es2 = next(iter(cluster2.engines.values()))
         # ensure responsive
         rc[:].apply_async(lambda: None).get(timeout=_timeout)
         if not sys.platform.startswith("win"):