forked from goodboy/tractor
1
0
Fork 0

Factor out msg unwrapping into a func

faster_daemon_cancels
Tyler Goodlet 2021-11-29 08:40:59 -05:00
parent 0e7234aa68
commit f6de7e0afd
1 changed files with 31 additions and 28 deletions

View File

@ -21,7 +21,7 @@ from .log import get_logger
from ._exceptions import ( from ._exceptions import (
unpack_error, unpack_error,
NoResult, NoResult,
RemoteActorError, # RemoteActorError,
ContextCancelled, ContextCancelled,
) )
from ._streaming import Context, ReceiveMsgStream from ._streaming import Context, ReceiveMsgStream
@ -54,8 +54,23 @@ def func_deats(func: Callable) -> Tuple[str, str]:
) )
def _unwrap_msg(
msg: dict[str, Any],
channel: Channel
) -> Any:
try:
return msg['return']
except KeyError:
# internal error should never get here
assert msg.get('cid'), "Received internal error at portal?"
raise unpack_error(msg, channel)
class Portal: class Portal:
"""A 'portal' to a(n) (remote) ``Actor``. '''
A 'portal' to a(n) (remote) ``Actor``.
A portal is "opened" (and eventually closed) by one side of an A portal is "opened" (and eventually closed) by one side of an
inter-actor communication context. The side which opens the portal inter-actor communication context. The side which opens the portal
@ -71,7 +86,7 @@ class Portal:
function calling semantics are supported transparently; hence it is function calling semantics are supported transparently; hence it is
like having a "portal" between the seperate actor memory spaces. like having a "portal" between the seperate actor memory spaces.
""" '''
def __init__(self, channel: Channel) -> None: def __init__(self, channel: Channel) -> None:
self.channel = channel self.channel = channel
# when this is set to a tuple returned from ``_submit()`` then # when this is set to a tuple returned from ``_submit()`` then
@ -130,16 +145,11 @@ class Portal:
resptype: str, resptype: str,
first_msg: dict first_msg: dict
) -> tuple[Any, dict[str, Any]]: ) -> dict[str, Any]:
assert resptype == 'asyncfunc' # single response
assert resptype == 'asyncfunc' # single response
msg = await recv_chan.receive() msg = await recv_chan.receive()
try: return msg
return msg['return'], msg
except KeyError:
# internal error should never get here
assert msg.get('cid'), "Received internal error at portal?"
raise unpack_error(msg, self.channel)
async def result(self) -> Any: async def result(self) -> Any:
"""Return the result(s) from the remote actor's "main" task. """Return the result(s) from the remote actor's "main" task.
@ -162,19 +172,9 @@ class Portal:
assert self._expect_result assert self._expect_result
if self._result_msg is None: if self._result_msg is None:
try: self._result_msg = await self._return_once(*self._expect_result)
result, self._result_msg = await self._return_once(
*self._expect_result)
except RemoteActorError as err:
result = err
else:
result = self._result_msg['return']
# re-raise error on every call return _unwrap_msg(self._result_msg, self.channel)
if isinstance(result, RemoteActorError):
raise result
return result
async def _cancel_streams(self): async def _cancel_streams(self):
# terminate all locally running async generator # terminate all locally running async generator
@ -249,10 +249,10 @@ class Portal:
instance methods in the remote runtime. Currently this should only instance methods in the remote runtime. Currently this should only
be used for `tractor` internals. be used for `tractor` internals.
""" """
value, _ = await self._return_once( msg = await self._return_once(
*(await self._submit(namespace_path, function_name, kwargs)) *(await self._submit(namespace_path, function_name, kwargs))
) )
return value return _unwrap_msg(msg, self.channel)
async def run( async def run(
self, self,
@ -293,9 +293,12 @@ class Portal:
fn_mod_path, fn_name = func_deats(func) fn_mod_path, fn_name = func_deats(func)
return (await self._return_once( return _unwrap_msg(
*(await self._submit(fn_mod_path, fn_name, kwargs)) await self._return_once(
))[0] *(await self._submit(fn_mod_path, fn_name, kwargs)),
),
self.channel,
)
@asynccontextmanager @asynccontextmanager
async def open_stream_from( async def open_stream_from(