From 74048b06a709d4193f7744b8b3819ded57671f7c Mon Sep 17 00:00:00 2001
From: Tyler Goodlet <jgbt@protonmail.com>
Date: Thu, 27 Jun 2024 16:25:46 -0400
Subject: [PATCH] Prep for legacy RPC API factor-n-remove
This change is adding commentary about the upcoming API removal and
simplification of nursery + portal internals; no actual code changes are
included.
The plan to (re)move the old RPC methods:
- `ActorNursery.run_in_actor()`
- `Portal.run()`
- `Portal.run_from_ns()`
and any related impl internals out of each conc-primitive and instead
into something like a `.hilevel.rpc` set of APIs which then are all
implemented using the newer and more lowlevel `Context`/`MsgStream`
primitives instead Bo
Further,
- formally deprecate the `Portal.result()` meth for
`.wait_for_result()`.
- only `log.info()` about runtime shutdown in the implicit root case.
---
tractor/_portal.py | 42 +++++++++++++++++++++++++++++++++++++----
tractor/_spawn.py | 4 ++--
tractor/_supervise.py | 44 +++++++++++++++++++++++++++++++------------
3 files changed, 72 insertions(+), 18 deletions(-)
diff --git a/tractor/_portal.py b/tractor/_portal.py
index 2c676e12..0f698836 100644
--- a/tractor/_portal.py
+++ b/tractor/_portal.py
@@ -121,7 +121,8 @@ class Portal:
)
return self.chan
- # TODO: factor this out into an `ActorNursery` wrapper
+ # TODO: factor this out into a `.highlevel` API-wrapper that uses
+ # a single `.open_context()` call underneath.
async def _submit_for_result(
self,
ns: str,
@@ -141,13 +142,22 @@ class Portal:
portal=self,
)
+ # TODO: we should deprecate this API right? since if we remove
+ # `.run_in_actor()` (and instead move it to a `.highlevel`
+ # wrapper api (around a single `.open_context()` call) we don't
+ # really have any notion of a "main" remote task any more?
+ #
# @api_frame
- async def result(self) -> Any:
+ async def wait_for_result(
+ self,
+ hide_tb: bool = True,
+ ) -> Any:
'''
- Return the result(s) from the remote actor's "main" task.
+ Return the final result delivered by a `Return`-msg from the
+ remote peer actor's "main" task's `return` statement.
'''
- __tracebackhide__ = True
+ __tracebackhide__: bool = hide_tb
# Check for non-rpc errors slapped on the
# channel for which we always raise
exc = self.channel._exc
@@ -182,6 +192,23 @@ class Portal:
return self._final_result_pld
+ # TODO: factor this out into a `.highlevel` API-wrapper that uses
+ # a single `.open_context()` call underneath.
+ async def result(
+ self,
+ *args,
+ **kwargs,
+ ) -> Any|Exception:
+ typname: str = type(self).__name__
+ log.warning(
+ f'`{typname}.result()` is DEPRECATED!\n'
+ f'Use `{typname}.wait_for_result()` instead!\n'
+ )
+ return await self.wait_for_result(
+ *args,
+ **kwargs,
+ )
+
async def _cancel_streams(self):
# terminate all locally running async generator
# IPC calls
@@ -240,6 +267,7 @@ class Portal:
f'{reminfo}'
)
+ # XXX the one spot we set it?
self.channel._cancel_called: bool = True
try:
# send cancel cmd - might not get response
@@ -279,6 +307,8 @@ class Portal:
)
return False
+ # TODO: do we still need this for low level `Actor`-runtime
+ # method calls or can we also remove it?
async def run_from_ns(
self,
namespace_path: str,
@@ -316,6 +346,8 @@ class Portal:
expect_msg=Return,
)
+ # TODO: factor this out into a `.highlevel` API-wrapper that uses
+ # a single `.open_context()` call underneath.
async def run(
self,
func: str,
@@ -370,6 +402,8 @@ class Portal:
expect_msg=Return,
)
+ # TODO: factor this out into a `.highlevel` API-wrapper that uses
+ # a single `.open_context()` call underneath.
@acm
async def open_stream_from(
self,
diff --git a/tractor/_spawn.py b/tractor/_spawn.py
index 481e2981..986c2e29 100644
--- a/tractor/_spawn.py
+++ b/tractor/_spawn.py
@@ -149,7 +149,7 @@ async def exhaust_portal(
# XXX: streams should never be reaped here since they should
# always be established and shutdown using a context manager api
- final: Any = await portal.result()
+ final: Any = await portal.wait_for_result()
except (
Exception,
@@ -223,8 +223,8 @@ async def cancel_on_completion(
async def hard_kill(
proc: trio.Process,
- terminate_after: int = 1.6,
+ terminate_after: int = 1.6,
# NOTE: for mucking with `.pause()`-ing inside the runtime
# whilst also hacking on it XD
# terminate_after: int = 99999,
diff --git a/tractor/_supervise.py b/tractor/_supervise.py
index 8f3574bb..fb737c12 100644
--- a/tractor/_supervise.py
+++ b/tractor/_supervise.py
@@ -80,6 +80,7 @@ class ActorNursery:
'''
def __init__(
self,
+ # TODO: maybe def these as fields of a struct looking type?
actor: Actor,
ria_nursery: trio.Nursery,
da_nursery: trio.Nursery,
@@ -88,8 +89,10 @@ class ActorNursery:
) -> None:
# self.supervisor = supervisor # TODO
self._actor: Actor = actor
- self._ria_nursery = ria_nursery
+
+ # TODO: rename to `._tn` for our conventional "task-nursery"
self._da_nursery = da_nursery
+
self._children: dict[
tuple[str, str],
tuple[
@@ -98,15 +101,13 @@ class ActorNursery:
Portal | None,
]
] = {}
- # portals spawned with ``run_in_actor()`` are
- # cancelled when their "main" result arrives
- self._cancel_after_result_on_exit: set = set()
+
self.cancelled: bool = False
self._join_procs = trio.Event()
self._at_least_one_child_in_debug: bool = False
self.errors = errors
- self.exited = trio.Event()
self._scope_error: BaseException|None = None
+ self.exited = trio.Event()
# NOTE: when no explicit call is made to
# `.open_root_actor()` by application code,
@@ -116,6 +117,13 @@ class ActorNursery:
# and syncing purposes to any actor opened nurseries.
self._implicit_runtime_started: bool = False
+ # TODO: remove the `.run_in_actor()` API and thus this 2ndary
+ # nursery when that API get's moved outside this primitive!
+ self._ria_nursery = ria_nursery
+ # portals spawned with ``run_in_actor()`` are
+ # cancelled when their "main" result arrives
+ self._cancel_after_result_on_exit: set = set()
+
async def start_actor(
self,
name: str,
@@ -126,10 +134,14 @@ class ActorNursery:
rpc_module_paths: list[str]|None = None,
enable_modules: list[str]|None = None,
loglevel: str|None = None, # set log level per subactor
- nursery: trio.Nursery|None = None,
debug_mode: bool|None = None,
infect_asyncio: bool = False,
+ # TODO: ideally we can rm this once we no longer have
+ # a `._ria_nursery` since the dependent APIs have been
+ # removed!
+ nursery: trio.Nursery|None = None,
+
) -> Portal:
'''
Start a (daemon) actor: an process that has no designated
@@ -200,6 +212,7 @@ class ActorNursery:
# |_ dynamic @context decoration on child side
# |_ implicit `Portal.open_context() as (ctx, first):`
# and `return first` on parent side.
+ # |_ mention how it's similar to `trio-parallel` API?
# -[ ] use @api_frame on the wrapper
async def run_in_actor(
self,
@@ -269,11 +282,14 @@ class ActorNursery:
) -> None:
'''
- Cancel this nursery by instructing each subactor to cancel
- itself and wait for all subactors to terminate.
+ Cancel this actor-nursery by instructing each subactor's
+ runtime to cancel and wait for all underlying sub-processes
+ to terminate.
- If ``hard_killl`` is set to ``True`` then kill the processes
- directly without any far end graceful ``trio`` cancellation.
+ If `hard_kill` is set then kill the processes directly using
+ the spawning-backend's API/OS-machinery without any attempt
+ at (graceful) `trio`-style cancellation using our
+ `Actor.cancel()`.
'''
__runtimeframe__: int = 1 # noqa
@@ -629,8 +645,12 @@ async def open_nursery(
f'|_{an}\n'
)
- # shutdown runtime if it was started
if implicit_runtime:
+ # shutdown runtime if it was started and report noisly
+ # that we're did so.
msg += '=> Shutting down actor runtime <=\n'
+ log.info(msg)
- log.info(msg)
+ else:
+ # keep noise low during std operation.
+ log.runtime(msg)