forked from goodboy/tractor
Compare commits
No commits in common. "ctx_cancel_semantics_and_overruns" and "master" have entirely different histories.
ctx_cancel
...
master
|
|
@ -49,7 +49,7 @@ async def worker(
|
|||
await ctx.started()
|
||||
|
||||
async with ctx.open_stream(
|
||||
allow_overruns=True,
|
||||
backpressure=True,
|
||||
) as stream:
|
||||
|
||||
# TODO: this with the below assert causes a hang bug?
|
||||
|
|
|
|||
|
|
@ -13,10 +13,7 @@ from typing import Optional
|
|||
import pytest
|
||||
import trio
|
||||
import tractor
|
||||
from tractor._exceptions import (
|
||||
StreamOverrun,
|
||||
ContextCancelled,
|
||||
)
|
||||
from tractor._exceptions import StreamOverrun
|
||||
|
||||
from conftest import tractor_test
|
||||
|
||||
|
|
@ -94,10 +91,7 @@ async def not_started_but_stream_opened(
|
|||
|
||||
@pytest.mark.parametrize(
|
||||
'target',
|
||||
[
|
||||
too_many_starteds,
|
||||
not_started_but_stream_opened,
|
||||
],
|
||||
[too_many_starteds, not_started_but_stream_opened],
|
||||
ids='misuse_type={}'.format,
|
||||
)
|
||||
def test_started_misuse(target):
|
||||
|
|
@ -234,88 +228,6 @@ def test_simple_context(
|
|||
trio.run(main)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'callee_returns_early',
|
||||
[True, False],
|
||||
ids=lambda item: f'callee_returns_early={item}'
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
'cancel_method',
|
||||
['ctx', 'portal'],
|
||||
ids=lambda item: f'cancel_method={item}'
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
'chk_ctx_result_before_exit',
|
||||
[True, False],
|
||||
ids=lambda item: f'chk_ctx_result_before_exit={item}'
|
||||
)
|
||||
def test_caller_cancels(
|
||||
cancel_method: str,
|
||||
chk_ctx_result_before_exit: bool,
|
||||
callee_returns_early: bool,
|
||||
):
|
||||
'''
|
||||
Verify that when the opening side of a context (aka the caller)
|
||||
cancels that context, the ctx does not raise a cancelled when
|
||||
either calling `.result()` or on context exit.
|
||||
|
||||
'''
|
||||
async def check_canceller(
|
||||
ctx: tractor.Context,
|
||||
) -> None:
|
||||
# should not raise yet return the remote
|
||||
# context cancelled error.
|
||||
res = await ctx.result()
|
||||
|
||||
if callee_returns_early:
|
||||
assert res == 'yo'
|
||||
|
||||
else:
|
||||
err = res
|
||||
assert isinstance(err, ContextCancelled)
|
||||
assert (
|
||||
tuple(err.canceller)
|
||||
==
|
||||
tractor.current_actor().uid
|
||||
)
|
||||
|
||||
async def main():
|
||||
async with tractor.open_nursery() as nursery:
|
||||
portal = await nursery.start_actor(
|
||||
'simple_context',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
timeout = 0.5 if not callee_returns_early else 2
|
||||
with trio.fail_after(timeout):
|
||||
async with portal.open_context(
|
||||
simple_setup_teardown,
|
||||
data=10,
|
||||
block_forever=not callee_returns_early,
|
||||
) as (ctx, sent):
|
||||
|
||||
if callee_returns_early:
|
||||
# ensure we block long enough before sending
|
||||
# a cancel such that the callee has already
|
||||
# returned it's result.
|
||||
await trio.sleep(0.5)
|
||||
|
||||
if cancel_method == 'ctx':
|
||||
await ctx.cancel()
|
||||
else:
|
||||
await portal.cancel_actor()
|
||||
|
||||
if chk_ctx_result_before_exit:
|
||||
await check_canceller(ctx)
|
||||
|
||||
if not chk_ctx_result_before_exit:
|
||||
await check_canceller(ctx)
|
||||
|
||||
if cancel_method != 'portal':
|
||||
await portal.cancel_actor()
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
# basic stream terminations:
|
||||
# - callee context closes without using stream
|
||||
# - caller context closes without using stream
|
||||
|
|
@ -594,6 +506,7 @@ async def test_callee_cancels_before_started():
|
|||
cancel_self,
|
||||
) as (ctx, sent):
|
||||
async with ctx.open_stream():
|
||||
|
||||
await trio.sleep_forever()
|
||||
|
||||
# raises a special cancel signal
|
||||
|
|
@ -646,6 +559,7 @@ async def keep_sending_from_callee(
|
|||
'overrun_by',
|
||||
[
|
||||
('caller', 1, never_open_stream),
|
||||
('cancel_caller_during_overrun', 1, never_open_stream),
|
||||
('callee', 0, keep_sending_from_callee),
|
||||
],
|
||||
ids='overrun_condition={}'.format,
|
||||
|
|
@ -675,13 +589,14 @@ def test_one_end_stream_not_opened(overrun_by):
|
|||
if 'caller' in overrunner:
|
||||
|
||||
async with ctx.open_stream() as stream:
|
||||
|
||||
# itersend +1 msg more then the buffer size
|
||||
# to cause the most basic overrun.
|
||||
for i in range(buf_size):
|
||||
print(f'sending {i}')
|
||||
await stream.send(i)
|
||||
|
||||
if 'cancel' in overrunner:
|
||||
# without this we block waiting on the child side
|
||||
await ctx.cancel()
|
||||
|
||||
else:
|
||||
# expect overrun error to be relayed back
|
||||
# and this sleep interrupted
|
||||
|
|
@ -695,9 +610,7 @@ def test_one_end_stream_not_opened(overrun_by):
|
|||
|
||||
# 2 overrun cases and the no overrun case (which pushes right up to
|
||||
# the msg limit)
|
||||
if (
|
||||
overrunner == 'caller'
|
||||
):
|
||||
if overrunner == 'caller' or 'cance' in overrunner:
|
||||
with pytest.raises(tractor.RemoteActorError) as excinfo:
|
||||
trio.run(main)
|
||||
|
||||
|
|
@ -721,102 +634,40 @@ async def echo_back_sequence(
|
|||
|
||||
ctx: tractor.Context,
|
||||
seq: list[int],
|
||||
wait_for_cancel: bool,
|
||||
allow_overruns_side: str,
|
||||
be_slow: bool = False,
|
||||
msg_buffer_size: int = 1,
|
||||
msg_buffer_size: Optional[int] = None,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Send endlessly on the calleee stream using a small buffer size
|
||||
setting on the contex to simulate backlogging that would normally
|
||||
cause overruns.
|
||||
Send endlessly on the calleee stream.
|
||||
|
||||
'''
|
||||
# NOTE: ensure that if the caller is expecting to cancel this task
|
||||
# that we stay echoing much longer then they are so we don't
|
||||
# return early instead of receive the cancel msg.
|
||||
total_batches: int = 1000 if wait_for_cancel else 6
|
||||
|
||||
await ctx.started()
|
||||
# await tractor.breakpoint()
|
||||
async with ctx.open_stream(
|
||||
msg_buffer_size=msg_buffer_size,
|
||||
|
||||
# literally the point of this test XD
|
||||
allow_overruns=(allow_overruns_side in {'child', 'both'}),
|
||||
) as stream:
|
||||
|
||||
# ensure mem chan settings are correct
|
||||
assert (
|
||||
ctx._send_chan._state.max_buffer_size
|
||||
==
|
||||
msg_buffer_size
|
||||
)
|
||||
|
||||
seq = list(seq) # bleh, msgpack sometimes ain't decoded right
|
||||
for _ in range(total_batches):
|
||||
seq = list(seq) # bleh, `msgpack`...
|
||||
count = 0
|
||||
while count < 3:
|
||||
batch = []
|
||||
async for msg in stream:
|
||||
batch.append(msg)
|
||||
if batch == seq:
|
||||
break
|
||||
|
||||
if be_slow:
|
||||
await trio.sleep(0.05)
|
||||
|
||||
print('callee waiting on next')
|
||||
|
||||
for msg in batch:
|
||||
print(f'callee sending {msg}')
|
||||
await stream.send(msg)
|
||||
|
||||
print(
|
||||
'EXITING CALLEEE:\n'
|
||||
f'{ctx.cancel_called_remote}'
|
||||
)
|
||||
return 'yo'
|
||||
count += 1
|
||||
|
||||
return 'yo'
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
# aka the side that will / should raise
|
||||
# and overrun under normal conditions.
|
||||
'allow_overruns_side',
|
||||
['parent', 'child', 'none', 'both'],
|
||||
ids=lambda item: f'allow_overruns_side={item}'
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
# aka the side that will / should raise
|
||||
# and overrun under normal conditions.
|
||||
'slow_side',
|
||||
['parent', 'child'],
|
||||
ids=lambda item: f'slow_side={item}'
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
'cancel_ctx',
|
||||
[True, False],
|
||||
ids=lambda item: f'cancel_ctx={item}'
|
||||
)
|
||||
def test_maybe_allow_overruns_stream(
|
||||
cancel_ctx: bool,
|
||||
slow_side: str,
|
||||
allow_overruns_side: str,
|
||||
loglevel: str,
|
||||
):
|
||||
def test_stream_backpressure():
|
||||
'''
|
||||
Demonstrate small overruns of each task back and forth
|
||||
on a stream not raising any errors by default by setting
|
||||
the ``allow_overruns=True``.
|
||||
|
||||
The original idea here was to show that if you set the feeder mem
|
||||
chan to a size smaller then the # of msgs sent you could could not
|
||||
get a `StreamOverrun` crash plus maybe get all the msgs that were
|
||||
sent. The problem with the "real backpressure" case is that due to
|
||||
the current arch it can result in the msg loop being blocked and thus
|
||||
blocking cancellation - which is like super bad. So instead this test
|
||||
had to be adjusted to more or less just "not send overrun errors" so
|
||||
as to handle the case where the sender just moreso cares about not getting
|
||||
errored out when it send to fast..
|
||||
on a stream not raising any errors by default.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
|
|
@ -824,104 +675,38 @@ def test_maybe_allow_overruns_stream(
|
|||
portal = await n.start_actor(
|
||||
'callee_sends_forever',
|
||||
enable_modules=[__name__],
|
||||
loglevel=loglevel,
|
||||
|
||||
# debug_mode=True,
|
||||
)
|
||||
seq = list(range(10))
|
||||
seq = list(range(3))
|
||||
async with portal.open_context(
|
||||
echo_back_sequence,
|
||||
seq=seq,
|
||||
wait_for_cancel=cancel_ctx,
|
||||
be_slow=(slow_side == 'child'),
|
||||
allow_overruns_side=allow_overruns_side,
|
||||
msg_buffer_size=1,
|
||||
) as (ctx, sent):
|
||||
|
||||
assert sent is None
|
||||
|
||||
async with ctx.open_stream(
|
||||
msg_buffer_size=1 if slow_side == 'parent' else None,
|
||||
allow_overruns=(allow_overruns_side in {'parent', 'both'}),
|
||||
) as stream:
|
||||
|
||||
total_batches: int = 2
|
||||
for _ in range(total_batches):
|
||||
async with ctx.open_stream(msg_buffer_size=1) as stream:
|
||||
count = 0
|
||||
while count < 3:
|
||||
for msg in seq:
|
||||
# print(f'root tx {msg}')
|
||||
print(f'caller sending {msg}')
|
||||
await stream.send(msg)
|
||||
if slow_side == 'parent':
|
||||
# NOTE: we make the parent slightly
|
||||
# slower, when it is slow, to make sure
|
||||
# that in the overruns everywhere case
|
||||
await trio.sleep(0.16)
|
||||
await trio.sleep(0.1)
|
||||
|
||||
batch = []
|
||||
async for msg in stream:
|
||||
print(f'root rx {msg}')
|
||||
batch.append(msg)
|
||||
if batch == seq:
|
||||
break
|
||||
|
||||
if cancel_ctx:
|
||||
# cancel the remote task
|
||||
print('sending root side cancel')
|
||||
await ctx.cancel()
|
||||
count += 1
|
||||
|
||||
res = await ctx.result()
|
||||
|
||||
if cancel_ctx:
|
||||
assert isinstance(res, ContextCancelled)
|
||||
assert tuple(res.canceller) == tractor.current_actor().uid
|
||||
|
||||
else:
|
||||
print(f'RX ROOT SIDE RESULT {res}')
|
||||
assert res == 'yo'
|
||||
# here the context should return
|
||||
assert await ctx.result() == 'yo'
|
||||
|
||||
# cancel the daemon
|
||||
await portal.cancel_actor()
|
||||
|
||||
if (
|
||||
allow_overruns_side == 'both'
|
||||
or slow_side == allow_overruns_side
|
||||
):
|
||||
trio.run(main)
|
||||
|
||||
elif (
|
||||
slow_side != allow_overruns_side
|
||||
):
|
||||
|
||||
with pytest.raises(tractor.RemoteActorError) as excinfo:
|
||||
trio.run(main)
|
||||
|
||||
err = excinfo.value
|
||||
|
||||
if (
|
||||
allow_overruns_side == 'none'
|
||||
):
|
||||
# depends on timing is is racy which side will
|
||||
# overrun first :sadkitty:
|
||||
|
||||
# NOTE: i tried to isolate to a deterministic case here
|
||||
# based on timeing, but i was kinda wasted, and i don't
|
||||
# think it's sane to catch them..
|
||||
assert err.type in (
|
||||
tractor.RemoteActorError,
|
||||
StreamOverrun,
|
||||
)
|
||||
|
||||
elif (
|
||||
slow_side == 'child'
|
||||
):
|
||||
assert err.type == StreamOverrun
|
||||
|
||||
elif slow_side == 'parent':
|
||||
assert err.type == tractor.RemoteActorError
|
||||
assert 'StreamOverrun' in err.msgdata['tb_str']
|
||||
|
||||
else:
|
||||
# if this hits the logic blocks from above are not
|
||||
# exhaustive..
|
||||
pytest.fail('PARAMETRIZED CASE GEN PROBLEM YO')
|
||||
trio.run(main)
|
||||
|
||||
|
||||
@tractor.context
|
||||
|
|
@ -952,18 +737,18 @@ async def attach_to_sleep_forever():
|
|||
finally:
|
||||
# XXX: previously this would trigger local
|
||||
# ``ContextCancelled`` to be received and raised in the
|
||||
# local context overriding any local error due to logic
|
||||
# inside ``_invoke()`` which checked for an error set on
|
||||
# ``Context._error`` and raised it in a cancellation
|
||||
# scenario.
|
||||
# ------
|
||||
# The problem is you can have a remote cancellation that
|
||||
# is part of a local error and we shouldn't raise
|
||||
# ``ContextCancelled`` **iff** we **were not** the side
|
||||
# of the context to initiate it, i.e.
|
||||
# local context overriding any local error due to
|
||||
# logic inside ``_invoke()`` which checked for
|
||||
# an error set on ``Context._error`` and raised it in
|
||||
# under a cancellation scenario.
|
||||
|
||||
# The problem is you can have a remote cancellation
|
||||
# that is part of a local error and we shouldn't raise
|
||||
# ``ContextCancelled`` **iff** we weren't the side of
|
||||
# the context to initiate it, i.e.
|
||||
# ``Context._cancel_called`` should **NOT** have been
|
||||
# set. The special logic to handle this case is now
|
||||
# inside ``Context._maybe_raise_from_remote_msg()`` XD
|
||||
# inside ``Context._may_raise_from_remote_msg()`` XD
|
||||
await peer_ctx.cancel()
|
||||
|
||||
|
||||
|
|
@ -984,10 +769,9 @@ async def error_before_started(
|
|||
|
||||
def test_do_not_swallow_error_before_started_by_remote_contextcancelled():
|
||||
'''
|
||||
Verify that an error raised in a remote context which itself opens
|
||||
another remote context, which it cancels, does not ovverride the
|
||||
original error that caused the cancellation of the secondardy
|
||||
context.
|
||||
Verify that an error raised in a remote context which itself opens another
|
||||
remote context, which it cancels, does not ovverride the original error that
|
||||
caused the cancellation of the secondardy context.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
|
|
|
|||
|
|
@ -15,7 +15,6 @@ import tractor
|
|||
from tractor import (
|
||||
to_asyncio,
|
||||
RemoteActorError,
|
||||
ContextCancelled,
|
||||
)
|
||||
from tractor.trionics import BroadcastReceiver
|
||||
|
||||
|
|
@ -225,23 +224,14 @@ def test_context_spawns_aio_task_that_errors(
|
|||
|
||||
await trio.sleep_forever()
|
||||
|
||||
return await ctx.result()
|
||||
with pytest.raises(RemoteActorError) as excinfo:
|
||||
trio.run(main)
|
||||
|
||||
err = excinfo.value
|
||||
assert isinstance(err, RemoteActorError)
|
||||
if parent_cancels:
|
||||
# bc the parent made the cancel request,
|
||||
# the error is not raised locally but instead
|
||||
# the context is exited silently
|
||||
res = trio.run(main)
|
||||
assert isinstance(res, ContextCancelled)
|
||||
assert 'root' in res.canceller[0]
|
||||
|
||||
assert err.type == trio.Cancelled
|
||||
else:
|
||||
expect = RemoteActorError
|
||||
with pytest.raises(expect) as excinfo:
|
||||
trio.run(main)
|
||||
|
||||
err = excinfo.value
|
||||
assert isinstance(err, expect)
|
||||
assert err.type == AssertionError
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -86,7 +86,7 @@ async def open_sequence_streamer(
|
|||
) as (ctx, first):
|
||||
|
||||
assert first is None
|
||||
async with ctx.open_stream(allow_overruns=True) as stream:
|
||||
async with ctx.open_stream(backpressure=True) as stream:
|
||||
yield stream
|
||||
|
||||
await portal.cancel_actor()
|
||||
|
|
@ -413,8 +413,8 @@ def test_ensure_slow_consumers_lag_out(
|
|||
seq = brx._state.subs[brx.key]
|
||||
assert seq == len(brx._state.queue) - 1
|
||||
|
||||
# all no_overruns entries in the underlying
|
||||
# channel should have been copied into the bcaster
|
||||
# all backpressured entries in the underlying
|
||||
# channel should have been copied into the caster
|
||||
# queue trailing-window
|
||||
async for i in rx:
|
||||
print(f'bped: {i}')
|
||||
|
|
|
|||
|
|
@ -15,20 +15,18 @@
|
|||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
"""
|
||||
tractor: structured concurrent ``trio``-"actors".
|
||||
tractor: structured concurrent "actors".
|
||||
|
||||
"""
|
||||
from exceptiongroup import BaseExceptionGroup
|
||||
|
||||
from ._clustering import open_actor_cluster
|
||||
from ._ipc import Channel
|
||||
from ._context import (
|
||||
Context,
|
||||
context,
|
||||
)
|
||||
from ._streaming import (
|
||||
Context,
|
||||
MsgStream,
|
||||
stream,
|
||||
context,
|
||||
)
|
||||
from ._discovery import (
|
||||
get_arbiter,
|
||||
|
|
|
|||
|
|
@ -1,778 +0,0 @@
|
|||
# tractor: structured concurrent "actors".
|
||||
# Copyright 2018-eternity Tyler Goodlet.
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
The fundamental cross process SC abstraction: an inter-actor,
|
||||
cancel-scope linked task "context".
|
||||
|
||||
A ``Context`` is very similar to the ``trio.Nursery.cancel_scope`` built
|
||||
into each ``trio.Nursery`` except it links the lifetimes of memory space
|
||||
disjoint, parallel executing tasks in separate actors.
|
||||
|
||||
'''
|
||||
from __future__ import annotations
|
||||
from collections import deque
|
||||
from contextlib import asynccontextmanager as acm
|
||||
from dataclasses import (
|
||||
dataclass,
|
||||
field,
|
||||
)
|
||||
from functools import partial
|
||||
import inspect
|
||||
from pprint import pformat
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
AsyncGenerator,
|
||||
TYPE_CHECKING,
|
||||
)
|
||||
import warnings
|
||||
|
||||
import trio
|
||||
|
||||
from ._exceptions import (
|
||||
unpack_error,
|
||||
pack_error,
|
||||
ContextCancelled,
|
||||
StreamOverrun,
|
||||
)
|
||||
from .log import get_logger
|
||||
from ._ipc import Channel
|
||||
from ._streaming import MsgStream
|
||||
from ._state import current_actor
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ._portal import Portal
|
||||
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Context:
|
||||
'''
|
||||
An inter-actor, ``trio``-task communication context.
|
||||
|
||||
NB: This class should never be instatiated directly, it is delivered
|
||||
by either,
|
||||
- runtime machinery to a remotely started task or,
|
||||
- by entering ``Portal.open_context()``.
|
||||
|
||||
and is always constructed using ``mkt_context()``.
|
||||
|
||||
Allows maintaining task or protocol specific state between
|
||||
2 communicating, parallel executing actor tasks. A unique context is
|
||||
allocated on each side of any task RPC-linked msg dialog, for
|
||||
every request to a remote actor from a portal. On the "callee"
|
||||
side a context is always allocated inside ``._runtime._invoke()``.
|
||||
|
||||
A context can be cancelled and (possibly eventually restarted) from
|
||||
either side of the underlying IPC channel, it can also open task
|
||||
oriented message streams, and acts more or less as an IPC aware
|
||||
inter-actor-task ``trio.CancelScope``.
|
||||
|
||||
'''
|
||||
chan: Channel
|
||||
cid: str
|
||||
|
||||
# these are the "feeder" channels for delivering
|
||||
# message values to the local task from the runtime
|
||||
# msg processing loop.
|
||||
_recv_chan: trio.MemoryReceiveChannel
|
||||
_send_chan: trio.MemorySendChannel
|
||||
|
||||
_remote_func_type: str | None = None
|
||||
|
||||
# only set on the caller side
|
||||
_portal: Portal | None = None # type: ignore # noqa
|
||||
_result: Any | int = None
|
||||
_remote_error: BaseException | None = None
|
||||
|
||||
# cancellation state
|
||||
_cancel_called: bool = False
|
||||
_cancelled_remote: tuple | None = None
|
||||
_cancel_msg: str | None = None
|
||||
_scope: trio.CancelScope | None = None
|
||||
_enter_debugger_on_cancel: bool = True
|
||||
|
||||
@property
|
||||
def cancel_called(self) -> bool:
|
||||
'''
|
||||
Records whether cancellation has been requested for this context
|
||||
by either an explicit call to ``.cancel()`` or an implicit call
|
||||
due to an error caught inside the ``Portal.open_context()``
|
||||
block.
|
||||
|
||||
'''
|
||||
return self._cancel_called
|
||||
|
||||
@property
|
||||
def cancel_called_remote(self) -> tuple[str, str] | None:
|
||||
'''
|
||||
``Actor.uid`` of the remote actor who's task was cancelled
|
||||
causing this side of the context to also be cancelled.
|
||||
|
||||
'''
|
||||
remote_uid = self._cancelled_remote
|
||||
if remote_uid:
|
||||
return tuple(remote_uid)
|
||||
|
||||
@property
|
||||
def cancelled_caught(self) -> bool:
|
||||
return self._scope.cancelled_caught
|
||||
|
||||
# init and streaming state
|
||||
_started_called: bool = False
|
||||
_started_received: bool = False
|
||||
_stream_opened: bool = False
|
||||
|
||||
# overrun handling machinery
|
||||
# NOTE: none of this provides "backpressure" to the remote
|
||||
# task, only an ability to not lose messages when the local
|
||||
# task is configured to NOT transmit ``StreamOverrun``s back
|
||||
# to the other side.
|
||||
_overflow_q: deque[dict] = field(
|
||||
default_factory=partial(
|
||||
deque,
|
||||
maxlen=616,
|
||||
)
|
||||
)
|
||||
_scope_nursery: trio.Nursery | None = None
|
||||
_in_overrun: bool = False
|
||||
_allow_overruns: bool = False
|
||||
|
||||
async def send_yield(
|
||||
self,
|
||||
data: Any,
|
||||
|
||||
) -> None:
|
||||
|
||||
warnings.warn(
|
||||
"`Context.send_yield()` is now deprecated. "
|
||||
"Use ``MessageStream.send()``. ",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
await self.chan.send({'yield': data, 'cid': self.cid})
|
||||
|
||||
async def send_stop(self) -> None:
|
||||
await self.chan.send({'stop': True, 'cid': self.cid})
|
||||
|
||||
async def _maybe_cancel_and_set_remote_error(
|
||||
self,
|
||||
error_msg: dict[str, Any],
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
(Maybe) unpack and raise a msg error into the local scope
|
||||
nursery for this context.
|
||||
|
||||
Acts as a form of "relay" for a remote error raised
|
||||
in the corresponding remote callee task.
|
||||
|
||||
'''
|
||||
# If this is an error message from a context opened by
|
||||
# ``Portal.open_context()`` we want to interrupt any ongoing
|
||||
# (child) tasks within that context to be notified of the remote
|
||||
# error relayed here.
|
||||
#
|
||||
# The reason we may want to raise the remote error immediately
|
||||
# is that there is no guarantee the associated local task(s)
|
||||
# will attempt to read from any locally opened stream any time
|
||||
# soon.
|
||||
#
|
||||
# NOTE: this only applies when
|
||||
# ``Portal.open_context()`` has been called since it is assumed
|
||||
# (currently) that other portal APIs (``Portal.run()``,
|
||||
# ``.run_in_actor()``) do their own error checking at the point
|
||||
# of the call and result processing.
|
||||
error = unpack_error(
|
||||
error_msg,
|
||||
self.chan,
|
||||
)
|
||||
|
||||
# XXX: set the remote side's error so that after we cancel
|
||||
# whatever task is the opener of this context it can raise
|
||||
# that error as the reason.
|
||||
self._remote_error = error
|
||||
|
||||
# always record the remote actor's uid since its cancellation
|
||||
# state is directly linked to ours (the local one).
|
||||
self._cancelled_remote = self.chan.uid
|
||||
|
||||
if (
|
||||
isinstance(error, ContextCancelled)
|
||||
):
|
||||
log.cancel(
|
||||
'Remote task-context sucessfully cancelled for '
|
||||
f'{self.chan.uid}:{self.cid}'
|
||||
)
|
||||
|
||||
if self._cancel_called:
|
||||
# from ._debug import breakpoint
|
||||
# await breakpoint()
|
||||
|
||||
# this is an expected cancel request response message
|
||||
# and we **don't need to raise it** in local cancel
|
||||
# scope since it will potentially override a real error.
|
||||
return
|
||||
else:
|
||||
log.error(
|
||||
f'Remote context error for {self.chan.uid}:{self.cid}:\n'
|
||||
f'{error_msg["error"]["tb_str"]}'
|
||||
)
|
||||
# TODO: tempted to **not** do this by-reraising in a
|
||||
# nursery and instead cancel a surrounding scope, detect
|
||||
# the cancellation, then lookup the error that was set?
|
||||
# YES! this is way better and simpler!
|
||||
if (
|
||||
self._scope
|
||||
):
|
||||
# from trio.testing import wait_all_tasks_blocked
|
||||
# await wait_all_tasks_blocked()
|
||||
# self._cancelled_remote = self.chan.uid
|
||||
self._scope.cancel()
|
||||
|
||||
# NOTE: this usage actually works here B)
|
||||
# from ._debug import breakpoint
|
||||
# await breakpoint()
|
||||
|
||||
# XXX: this will break early callee results sending
|
||||
# since when `.result()` is finally called, this
|
||||
# chan will be closed..
|
||||
# if self._recv_chan:
|
||||
# await self._recv_chan.aclose()
|
||||
|
||||
async def cancel(
|
||||
self,
|
||||
msg: str | None = None,
|
||||
timeout: float = 0.616,
|
||||
# timeout: float = 1000,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Cancel this inter-actor-task context.
|
||||
|
||||
Request that the far side cancel it's current linked context,
|
||||
Timeout quickly in an attempt to sidestep 2-generals...
|
||||
|
||||
'''
|
||||
side = 'caller' if self._portal else 'callee'
|
||||
if msg:
|
||||
assert side == 'callee', 'Only callee side can provide cancel msg'
|
||||
|
||||
log.cancel(f'Cancelling {side} side of context to {self.chan.uid}')
|
||||
|
||||
self._cancel_called = True
|
||||
# await _debug.breakpoint()
|
||||
# breakpoint()
|
||||
|
||||
if side == 'caller':
|
||||
if not self._portal:
|
||||
raise RuntimeError(
|
||||
"No portal found, this is likely a callee side context"
|
||||
)
|
||||
|
||||
cid = self.cid
|
||||
with trio.move_on_after(timeout) as cs:
|
||||
cs.shield = True
|
||||
log.cancel(
|
||||
f"Cancelling stream {cid} to "
|
||||
f"{self._portal.channel.uid}")
|
||||
|
||||
# NOTE: we're telling the far end actor to cancel a task
|
||||
# corresponding to *this actor*. The far end local channel
|
||||
# instance is passed to `Actor._cancel_task()` implicitly.
|
||||
await self._portal.run_from_ns(
|
||||
'self',
|
||||
'_cancel_task',
|
||||
cid=cid,
|
||||
)
|
||||
# print("EXITING CANCEL CALL")
|
||||
|
||||
if cs.cancelled_caught:
|
||||
# XXX: there's no way to know if the remote task was indeed
|
||||
# cancelled in the case where the connection is broken or
|
||||
# some other network error occurred.
|
||||
# if not self._portal.channel.connected():
|
||||
if not self.chan.connected():
|
||||
log.cancel(
|
||||
"May have failed to cancel remote task "
|
||||
f"{cid} for {self._portal.channel.uid}")
|
||||
else:
|
||||
log.cancel(
|
||||
"Timed out on cancelling remote task "
|
||||
f"{cid} for {self._portal.channel.uid}")
|
||||
|
||||
# callee side remote task
|
||||
else:
|
||||
self._cancel_msg = msg
|
||||
|
||||
# TODO: should we have an explicit cancel message
|
||||
# or is relaying the local `trio.Cancelled` as an
|
||||
# {'error': trio.Cancelled, cid: "blah"} enough?
|
||||
# This probably gets into the discussion in
|
||||
# https://github.com/goodboy/tractor/issues/36
|
||||
assert self._scope
|
||||
self._scope.cancel()
|
||||
|
||||
@acm
|
||||
async def open_stream(
|
||||
|
||||
self,
|
||||
allow_overruns: bool | None = False,
|
||||
msg_buffer_size: int | None = None,
|
||||
|
||||
) -> AsyncGenerator[MsgStream, None]:
|
||||
'''
|
||||
Open a ``MsgStream``, a bi-directional stream connected to the
|
||||
cross-actor (far end) task for this ``Context``.
|
||||
|
||||
This context manager must be entered on both the caller and
|
||||
callee for the stream to logically be considered "connected".
|
||||
|
||||
A ``MsgStream`` is currently "one-shot" use, meaning if you
|
||||
close it you can not "re-open" it for streaming and instead you
|
||||
must re-establish a new surrounding ``Context`` using
|
||||
``Portal.open_context()``. In the future this may change but
|
||||
currently there seems to be no obvious reason to support
|
||||
"re-opening":
|
||||
- pausing a stream can be done with a message.
|
||||
- task errors will normally require a restart of the entire
|
||||
scope of the inter-actor task context due to the nature of
|
||||
``trio``'s cancellation system.
|
||||
|
||||
'''
|
||||
actor = current_actor()
|
||||
|
||||
# here we create a mem chan that corresponds to the
|
||||
# far end caller / callee.
|
||||
|
||||
# Likewise if the surrounding context has been cancelled we error here
|
||||
# since it likely means the surrounding block was exited or
|
||||
# killed
|
||||
|
||||
if self._cancel_called:
|
||||
task = trio.lowlevel.current_task().name
|
||||
raise ContextCancelled(
|
||||
f'Context around {actor.uid[0]}:{task} was already cancelled!'
|
||||
)
|
||||
|
||||
if not self._portal and not self._started_called:
|
||||
raise RuntimeError(
|
||||
'Context.started()` must be called before opening a stream'
|
||||
)
|
||||
|
||||
# NOTE: in one way streaming this only happens on the
|
||||
# caller side inside `Actor.start_remote_task()` so if you try
|
||||
# to send a stop from the caller to the callee in the
|
||||
# single-direction-stream case you'll get a lookup error
|
||||
# currently.
|
||||
ctx = actor.get_context(
|
||||
self.chan,
|
||||
self.cid,
|
||||
msg_buffer_size=msg_buffer_size,
|
||||
allow_overruns=allow_overruns,
|
||||
)
|
||||
ctx._allow_overruns = allow_overruns
|
||||
assert ctx is self
|
||||
|
||||
# XXX: If the underlying channel feeder receive mem chan has
|
||||
# been closed then likely client code has already exited
|
||||
# a ``.open_stream()`` block prior or there was some other
|
||||
# unanticipated error or cancellation from ``trio``.
|
||||
|
||||
if ctx._recv_chan._closed:
|
||||
raise trio.ClosedResourceError(
|
||||
'The underlying channel for this stream was already closed!?')
|
||||
|
||||
async with MsgStream(
|
||||
ctx=self,
|
||||
rx_chan=ctx._recv_chan,
|
||||
) as stream:
|
||||
|
||||
if self._portal:
|
||||
self._portal._streams.add(stream)
|
||||
|
||||
try:
|
||||
self._stream_opened = True
|
||||
|
||||
# XXX: do we need this?
|
||||
# ensure we aren't cancelled before yielding the stream
|
||||
# await trio.lowlevel.checkpoint()
|
||||
yield stream
|
||||
|
||||
# NOTE: Make the stream "one-shot use". On exit, signal
|
||||
# ``trio.EndOfChannel``/``StopAsyncIteration`` to the
|
||||
# far end.
|
||||
await stream.aclose()
|
||||
|
||||
finally:
|
||||
if self._portal:
|
||||
try:
|
||||
self._portal._streams.remove(stream)
|
||||
except KeyError:
|
||||
log.warning(
|
||||
f'Stream was already destroyed?\n'
|
||||
f'actor: {self.chan.uid}\n'
|
||||
f'ctx id: {self.cid}'
|
||||
)
|
||||
|
||||
def _maybe_raise_remote_err(
|
||||
self,
|
||||
err: Exception,
|
||||
) -> None:
|
||||
# NOTE: whenever the context's "opener" side (task) **is**
|
||||
# the side which requested the cancellation (likekly via
|
||||
# ``Context.cancel()``), we don't want to re-raise that
|
||||
# cancellation signal locally (would be akin to
|
||||
# a ``trio.Nursery`` nursery raising ``trio.Cancelled``
|
||||
# whenever ``CancelScope.cancel()`` was called) and instead
|
||||
# silently reap the expected cancellation "error"-msg.
|
||||
# if 'pikerd' in err.msgdata['tb_str']:
|
||||
# # from . import _debug
|
||||
# # await _debug.breakpoint()
|
||||
# breakpoint()
|
||||
|
||||
if (
|
||||
isinstance(err, ContextCancelled)
|
||||
and (
|
||||
self._cancel_called
|
||||
or self.chan._cancel_called
|
||||
or tuple(err.canceller) == current_actor().uid
|
||||
)
|
||||
):
|
||||
return err
|
||||
|
||||
raise err # from None
|
||||
|
||||
async def result(self) -> Any | Exception:
|
||||
'''
|
||||
From some (caller) side task, wait for and return the final
|
||||
result from the remote (callee) side's task.
|
||||
|
||||
This provides a mechanism for one task running in some actor to wait
|
||||
on another task at the other side, in some other actor, to terminate.
|
||||
|
||||
If the remote task is still in a streaming state (it is delivering
|
||||
values from inside a ``Context.open_stream():`` block, then those
|
||||
msgs are drained but discarded since it is presumed this side of
|
||||
the context has already finished with its own streaming logic.
|
||||
|
||||
If the remote context (or its containing actor runtime) was
|
||||
canceled, either by a local task calling one of
|
||||
``Context.cancel()`` or `Portal.cancel_actor()``, we ignore the
|
||||
received ``ContextCancelled`` exception if the context or
|
||||
underlying IPC channel is marked as having been "cancel called".
|
||||
This is similar behavior to using ``trio.Nursery.cancel()``
|
||||
wherein tasks which raise ``trio.Cancel`` are silently reaped;
|
||||
the main different in this API is in the "cancel called" case,
|
||||
instead of just not raising, we also return the exception *as
|
||||
the result* since client code may be interested in the details
|
||||
of the remote cancellation.
|
||||
|
||||
'''
|
||||
assert self._portal, "Context.result() can not be called from callee!"
|
||||
assert self._recv_chan
|
||||
|
||||
# from . import _debug
|
||||
# await _debug.breakpoint()
|
||||
|
||||
re = self._remote_error
|
||||
if re:
|
||||
self._maybe_raise_remote_err(re)
|
||||
return re
|
||||
|
||||
if (
|
||||
self._result == id(self)
|
||||
and not self._remote_error
|
||||
and not self._recv_chan._closed # type: ignore
|
||||
):
|
||||
# wait for a final context result consuming
|
||||
# and discarding any bi dir stream msgs still
|
||||
# in transit from the far end.
|
||||
while True:
|
||||
msg = await self._recv_chan.receive()
|
||||
try:
|
||||
self._result = msg['return']
|
||||
|
||||
# NOTE: we don't need to do this right?
|
||||
# XXX: only close the rx mem chan AFTER
|
||||
# a final result is retreived.
|
||||
# if self._recv_chan:
|
||||
# await self._recv_chan.aclose()
|
||||
|
||||
break
|
||||
except KeyError: # as msgerr:
|
||||
|
||||
if 'yield' in msg:
|
||||
# far end task is still streaming to us so discard
|
||||
log.warning(f'Discarding stream delivered {msg}')
|
||||
continue
|
||||
|
||||
elif 'stop' in msg:
|
||||
log.debug('Remote stream terminated')
|
||||
continue
|
||||
|
||||
# internal error should never get here
|
||||
assert msg.get('cid'), (
|
||||
"Received internal error at portal?")
|
||||
|
||||
err = unpack_error(
|
||||
msg,
|
||||
self._portal.channel
|
||||
) # from msgerr
|
||||
|
||||
err = self._maybe_raise_remote_err(err)
|
||||
self._remote_err = err
|
||||
|
||||
return self._remote_error or self._result
|
||||
|
||||
async def started(
|
||||
self,
|
||||
value: Any | None = None
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Indicate to calling actor's task that this linked context
|
||||
has started and send ``value`` to the other side.
|
||||
|
||||
On the calling side ``value`` is the second item delivered
|
||||
in the tuple returned by ``Portal.open_context()``.
|
||||
|
||||
'''
|
||||
if self._portal:
|
||||
raise RuntimeError(
|
||||
f"Caller side context {self} can not call started!")
|
||||
|
||||
elif self._started_called:
|
||||
raise RuntimeError(
|
||||
f"called 'started' twice on context with {self.chan.uid}")
|
||||
|
||||
await self.chan.send({'started': value, 'cid': self.cid})
|
||||
self._started_called = True
|
||||
|
||||
# TODO: do we need a restart api?
|
||||
# async def restart(self) -> None:
|
||||
# pass
|
||||
|
||||
async def _drain_overflows(
|
||||
self,
|
||||
) -> None:
|
||||
'''
|
||||
Private task spawned to push newly received msgs to the local
|
||||
task which getting overrun by the remote side.
|
||||
|
||||
In order to not block the rpc msg loop, but also not discard
|
||||
msgs received in this context, we need to async push msgs in
|
||||
a new task which only runs for as long as the local task is in
|
||||
an overrun state.
|
||||
|
||||
'''
|
||||
self._in_overrun = True
|
||||
try:
|
||||
while self._overflow_q:
|
||||
# NOTE: these msgs should never be errors since we always do
|
||||
# the check prior to checking if we're in an overrun state
|
||||
# inside ``.deliver_msg()``.
|
||||
msg = self._overflow_q.popleft()
|
||||
try:
|
||||
await self._send_chan.send(msg)
|
||||
except trio.BrokenResourceError:
|
||||
log.warning(
|
||||
f"{self._send_chan} consumer is already closed"
|
||||
)
|
||||
return
|
||||
except trio.Cancelled:
|
||||
# we are obviously still in overrun
|
||||
# but the context is being closed anyway
|
||||
# so we just warn that there are un received
|
||||
# msgs still..
|
||||
self._overflow_q.appendleft(msg)
|
||||
fmt_msgs = ''
|
||||
for msg in self._overflow_q:
|
||||
fmt_msgs += f'{pformat(msg)}\n'
|
||||
|
||||
log.warning(
|
||||
f'Context for {self.cid} is being closed while '
|
||||
'in an overrun state!\n'
|
||||
'Discarding the following msgs:\n'
|
||||
f'{fmt_msgs}\n'
|
||||
)
|
||||
raise
|
||||
|
||||
finally:
|
||||
# task is now finished with the backlog so mark us as
|
||||
# no longer in backlog.
|
||||
self._in_overrun = False
|
||||
|
||||
async def _deliver_msg(
|
||||
self,
|
||||
msg: dict,
|
||||
|
||||
draining: bool = False,
|
||||
|
||||
) -> bool:
|
||||
|
||||
cid = self.cid
|
||||
chan = self.chan
|
||||
uid = chan.uid
|
||||
send_chan: trio.MemorySendChannel = self._send_chan
|
||||
|
||||
log.runtime(
|
||||
f"Delivering {msg} from {uid} to caller {cid}"
|
||||
)
|
||||
|
||||
error = msg.get('error')
|
||||
if error:
|
||||
await self._maybe_cancel_and_set_remote_error(msg)
|
||||
|
||||
if (
|
||||
self._in_overrun
|
||||
):
|
||||
self._overflow_q.append(msg)
|
||||
return False
|
||||
|
||||
try:
|
||||
send_chan.send_nowait(msg)
|
||||
return True
|
||||
# if an error is deteced we should always
|
||||
# expect it to be raised by any context (stream)
|
||||
# consumer task
|
||||
|
||||
except trio.BrokenResourceError:
|
||||
# TODO: what is the right way to handle the case where the
|
||||
# local task has already sent a 'stop' / StopAsyncInteration
|
||||
# to the other side but and possibly has closed the local
|
||||
# feeder mem chan? Do we wait for some kind of ack or just
|
||||
# let this fail silently and bubble up (currently)?
|
||||
|
||||
# XXX: local consumer has closed their side
|
||||
# so cancel the far end streaming task
|
||||
log.warning(f"{send_chan} consumer is already closed")
|
||||
return False
|
||||
|
||||
# NOTE XXX: by default we do **not** maintain context-stream
|
||||
# backpressure and instead opt to relay stream overrun errors to
|
||||
# the sender; the main motivation is that using bp can block the
|
||||
# msg handling loop which calls into this method!
|
||||
except trio.WouldBlock:
|
||||
# XXX: always push an error even if the local
|
||||
# receiver is in overrun state.
|
||||
# await self._maybe_cancel_and_set_remote_error(msg)
|
||||
|
||||
local_uid = current_actor().uid
|
||||
lines = [
|
||||
f'OVERRUN on actor-task context {cid}@{local_uid}!\n'
|
||||
# TODO: put remote task name here if possible?
|
||||
f'remote sender actor: {uid}',
|
||||
# TODO: put task func name here and maybe an arrow
|
||||
# from sender to overrunner?
|
||||
# f'local task {self.func_name}'
|
||||
]
|
||||
if not self._stream_opened:
|
||||
lines.insert(
|
||||
1,
|
||||
f'\n*** No stream open on `{local_uid[0]}` side! ***\n'
|
||||
)
|
||||
|
||||
text = '\n'.join(lines)
|
||||
|
||||
# XXX: lul, this really can't be backpressure since any
|
||||
# blocking here will block the entire msg loop rpc sched for
|
||||
# a whole channel.. maybe we should rename it?
|
||||
if self._allow_overruns:
|
||||
text += f'\nStarting overflow queuing task on msg: {msg}'
|
||||
log.warning(text)
|
||||
if (
|
||||
not self._in_overrun
|
||||
):
|
||||
self._overflow_q.append(msg)
|
||||
n = self._scope_nursery
|
||||
assert not n.child_tasks
|
||||
try:
|
||||
n.start_soon(
|
||||
self._drain_overflows,
|
||||
)
|
||||
except RuntimeError:
|
||||
# if the nursery is already cancelled due to
|
||||
# this context exiting or in error, we ignore
|
||||
# the nursery error since we never expected
|
||||
# anything different.
|
||||
return False
|
||||
else:
|
||||
try:
|
||||
raise StreamOverrun(text)
|
||||
except StreamOverrun as err:
|
||||
err_msg = pack_error(err)
|
||||
err_msg['cid'] = cid
|
||||
try:
|
||||
await chan.send(err_msg)
|
||||
except trio.BrokenResourceError:
|
||||
# XXX: local consumer has closed their side
|
||||
# so cancel the far end streaming task
|
||||
log.warning(f"{chan} is already closed")
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def mk_context(
|
||||
chan: Channel,
|
||||
cid: str,
|
||||
msg_buffer_size: int = 2**6,
|
||||
|
||||
**kwargs,
|
||||
|
||||
) -> Context:
|
||||
'''
|
||||
Internal factory to create an inter-actor task ``Context``.
|
||||
|
||||
This is called by internals and should generally never be called
|
||||
by user code.
|
||||
|
||||
'''
|
||||
send_chan: trio.MemorySendChannel
|
||||
recv_chan: trio.MemoryReceiveChannel
|
||||
send_chan, recv_chan = trio.open_memory_channel(msg_buffer_size)
|
||||
|
||||
ctx = Context(
|
||||
chan,
|
||||
cid,
|
||||
_send_chan=send_chan,
|
||||
_recv_chan=recv_chan,
|
||||
**kwargs,
|
||||
)
|
||||
ctx._result: int | Any = id(ctx)
|
||||
return ctx
|
||||
|
||||
|
||||
def context(func: Callable) -> Callable:
|
||||
'''
|
||||
Mark an async function as a streaming routine with ``@context``.
|
||||
|
||||
'''
|
||||
# TODO: apply whatever solution ``mypy`` ends up picking for this:
|
||||
# https://github.com/python/mypy/issues/2087#issuecomment-769266912
|
||||
func._tractor_context_function = True # type: ignore
|
||||
|
||||
sig = inspect.signature(func)
|
||||
params = sig.parameters
|
||||
if 'ctx' not in params:
|
||||
raise TypeError(
|
||||
"The first argument to the context function "
|
||||
f"{func.__name__} must be `ctx: tractor.Context`"
|
||||
)
|
||||
return func
|
||||
|
|
@ -41,10 +41,8 @@ async def get_arbiter(
|
|||
port: int,
|
||||
|
||||
) -> AsyncGenerator[Union[Portal, LocalPortal], None]:
|
||||
'''
|
||||
Return a portal instance connected to a local or remote
|
||||
'''Return a portal instance connected to a local or remote
|
||||
arbiter.
|
||||
|
||||
'''
|
||||
actor = current_actor()
|
||||
|
||||
|
|
@ -136,16 +134,12 @@ async def find_actor(
|
|||
@acm
|
||||
async def wait_for_actor(
|
||||
name: str,
|
||||
arbiter_sockaddr: tuple[str, int] | None = None,
|
||||
# registry_addr: tuple[str, int] | None = None,
|
||||
|
||||
arbiter_sockaddr: tuple[str, int] | None = None
|
||||
) -> AsyncGenerator[Portal, None]:
|
||||
'''
|
||||
Wait on an actor to register with the arbiter.
|
||||
"""Wait on an actor to register with the arbiter.
|
||||
|
||||
A portal to the first registered actor is returned.
|
||||
|
||||
'''
|
||||
"""
|
||||
actor = current_actor()
|
||||
|
||||
async with get_arbiter(
|
||||
|
|
|
|||
|
|
@ -132,7 +132,7 @@ def _trio_main(
|
|||
else:
|
||||
trio.run(trio_main)
|
||||
except KeyboardInterrupt:
|
||||
log.cancel(f"Actor {actor.uid} received KBI")
|
||||
log.warning(f"Actor {actor.uid} received KBI")
|
||||
|
||||
finally:
|
||||
log.info(f"Actor {actor.uid} terminated")
|
||||
|
|
|
|||
|
|
@ -18,18 +18,18 @@
|
|||
Our classy exception set.
|
||||
|
||||
"""
|
||||
import builtins
|
||||
import importlib
|
||||
from typing import (
|
||||
Any,
|
||||
Optional,
|
||||
Type,
|
||||
)
|
||||
import importlib
|
||||
import builtins
|
||||
import traceback
|
||||
|
||||
import exceptiongroup as eg
|
||||
import trio
|
||||
|
||||
from ._state import current_actor
|
||||
|
||||
_this_mod = importlib.import_module(__name__)
|
||||
|
||||
|
|
@ -44,7 +44,7 @@ class RemoteActorError(Exception):
|
|||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
suberror_type: Type[BaseException] | None = None,
|
||||
suberror_type: Optional[Type[BaseException]] = None,
|
||||
**msgdata
|
||||
|
||||
) -> None:
|
||||
|
|
@ -53,36 +53,21 @@ class RemoteActorError(Exception):
|
|||
self.type = suberror_type
|
||||
self.msgdata = msgdata
|
||||
|
||||
@property
|
||||
def src_actor_uid(self) -> tuple[str, str] | None:
|
||||
return self.msgdata.get('src_actor_uid')
|
||||
|
||||
|
||||
class InternalActorError(RemoteActorError):
|
||||
'''
|
||||
Remote internal ``tractor`` error indicating
|
||||
"""Remote internal ``tractor`` error indicating
|
||||
failure of some primitive or machinery.
|
||||
|
||||
'''
|
||||
|
||||
|
||||
class ContextCancelled(RemoteActorError):
|
||||
'''
|
||||
Inter-actor task context was cancelled by either a call to
|
||||
``Portal.cancel_actor()`` or ``Context.cancel()``.
|
||||
|
||||
'''
|
||||
@property
|
||||
def canceller(self) -> tuple[str, str] | None:
|
||||
value = self.msgdata.get('canceller')
|
||||
if value:
|
||||
return tuple(value)
|
||||
"""
|
||||
|
||||
|
||||
class TransportClosed(trio.ClosedResourceError):
|
||||
"Underlying channel transport was closed prior to use"
|
||||
|
||||
|
||||
class ContextCancelled(RemoteActorError):
|
||||
"Inter-actor task context cancelled itself on the callee side."
|
||||
|
||||
|
||||
class NoResult(RuntimeError):
|
||||
"No final result is expected for this actor"
|
||||
|
||||
|
|
@ -121,17 +106,13 @@ def pack_error(
|
|||
else:
|
||||
tb_str = traceback.format_exc()
|
||||
|
||||
error_msg = {
|
||||
'tb_str': tb_str,
|
||||
'type_str': type(exc).__name__,
|
||||
'src_actor_uid': current_actor().uid,
|
||||
return {
|
||||
'error': {
|
||||
'tb_str': tb_str,
|
||||
'type_str': type(exc).__name__,
|
||||
}
|
||||
}
|
||||
|
||||
if isinstance(exc, ContextCancelled):
|
||||
error_msg.update(exc.msgdata)
|
||||
|
||||
return {'error': error_msg}
|
||||
|
||||
|
||||
def unpack_error(
|
||||
|
||||
|
|
@ -155,7 +136,7 @@ def unpack_error(
|
|||
|
||||
if type_name == 'ContextCancelled':
|
||||
err_type = ContextCancelled
|
||||
suberror_type = RemoteActorError
|
||||
suberror_type = trio.Cancelled
|
||||
|
||||
else: # try to lookup a suitable local error type
|
||||
for ns in [
|
||||
|
|
|
|||
|
|
@ -45,8 +45,10 @@ from ._exceptions import (
|
|||
NoResult,
|
||||
ContextCancelled,
|
||||
)
|
||||
from ._context import Context
|
||||
from ._streaming import MsgStream
|
||||
from ._streaming import (
|
||||
Context,
|
||||
MsgStream,
|
||||
)
|
||||
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
|
@ -101,7 +103,7 @@ class Portal:
|
|||
# When set to a ``Context`` (when _submit_for_result is called)
|
||||
# it is expected that ``result()`` will be awaited at some
|
||||
# point.
|
||||
self._expect_result: Context | None = None
|
||||
self._expect_result: Optional[Context] = None
|
||||
self._streams: set[MsgStream] = set()
|
||||
self.actor = current_actor()
|
||||
|
||||
|
|
@ -207,10 +209,7 @@ class Portal:
|
|||
try:
|
||||
# send cancel cmd - might not get response
|
||||
# XXX: sure would be nice to make this work with a proper shield
|
||||
with trio.move_on_after(
|
||||
timeout
|
||||
or self.cancel_timeout
|
||||
) as cs:
|
||||
with trio.move_on_after(timeout or self.cancel_timeout) as cs:
|
||||
cs.shield = True
|
||||
|
||||
await self.run_from_ns('self', 'cancel')
|
||||
|
|
@ -331,9 +330,7 @@ class Portal:
|
|||
f'{async_gen_func} must be an async generator function!')
|
||||
|
||||
fn_mod_path, fn_name = NamespacePath.from_ref(
|
||||
async_gen_func
|
||||
).to_tuple()
|
||||
|
||||
async_gen_func).to_tuple()
|
||||
ctx = await self.actor.start_remote_task(
|
||||
self.channel,
|
||||
fn_mod_path,
|
||||
|
|
@ -380,7 +377,6 @@ class Portal:
|
|||
|
||||
self,
|
||||
func: Callable,
|
||||
allow_overruns: bool = False,
|
||||
**kwargs,
|
||||
|
||||
) -> AsyncGenerator[tuple[Context, Any], None]:
|
||||
|
|
@ -400,26 +396,13 @@ class Portal:
|
|||
raise TypeError(
|
||||
f'{func} must be an async generator function!')
|
||||
|
||||
# TODO: i think from here onward should probably
|
||||
# just be factored into an `@acm` inside a new
|
||||
# a new `_context.py` mod.
|
||||
fn_mod_path, fn_name = NamespacePath.from_ref(func).to_tuple()
|
||||
|
||||
ctx = await self.actor.start_remote_task(
|
||||
self.channel,
|
||||
fn_mod_path,
|
||||
fn_name,
|
||||
kwargs,
|
||||
|
||||
# NOTE: it's imporant to expose this since you might
|
||||
# get the case where the parent who opened the context does
|
||||
# not open a stream until after some slow startup/init
|
||||
# period, in which case when the first msg is read from
|
||||
# the feeder mem chan, say when first calling
|
||||
# `Context.open_stream(allow_overruns=True)`, the overrun condition will be
|
||||
# raised before any ignoring of overflow msgs can take
|
||||
# place..
|
||||
allow_overruns=allow_overruns,
|
||||
kwargs
|
||||
)
|
||||
|
||||
assert ctx._remote_func_type == 'context'
|
||||
|
|
@ -443,47 +426,29 @@ class Portal:
|
|||
f' but received a non-error msg:\n{pformat(msg)}'
|
||||
)
|
||||
|
||||
_err: BaseException | None = None
|
||||
ctx._portal: Portal = self
|
||||
_err: Optional[BaseException] = None
|
||||
ctx._portal = self
|
||||
|
||||
uid: tuple = self.channel.uid
|
||||
cid: str = ctx.cid
|
||||
etype: Type[BaseException] | None = None
|
||||
uid = self.channel.uid
|
||||
cid = ctx.cid
|
||||
etype: Optional[Type[BaseException]] = None
|
||||
|
||||
# deliver context instance and .started() msg value in enter
|
||||
# tuple.
|
||||
# deliver context instance and .started() msg value in open tuple.
|
||||
try:
|
||||
async with trio.open_nursery() as nurse:
|
||||
ctx._scope_nursery = nurse
|
||||
ctx._scope = nurse.cancel_scope
|
||||
async with trio.open_nursery() as scope_nursery:
|
||||
ctx._scope_nursery = scope_nursery
|
||||
|
||||
# do we need this?
|
||||
# await trio.lowlevel.checkpoint()
|
||||
|
||||
yield ctx, first
|
||||
|
||||
# when in allow_ovveruns mode there may be lingering
|
||||
# overflow sender tasks remaining?
|
||||
if nurse.child_tasks:
|
||||
# ensure we are in overrun state with
|
||||
# ``._allow_overruns=True`` bc otherwise
|
||||
# there should be no tasks in this nursery!
|
||||
if (
|
||||
not ctx._allow_overruns
|
||||
or len(nurse.child_tasks) > 1
|
||||
):
|
||||
raise RuntimeError(
|
||||
'Context has sub-tasks but is '
|
||||
'not in `allow_overruns=True` Mode!?'
|
||||
)
|
||||
ctx._scope.cancel()
|
||||
|
||||
except ContextCancelled as err:
|
||||
_err = err
|
||||
|
||||
# swallow and mask cross-actor task context cancels that
|
||||
# were initiated by *this* side's task.
|
||||
if not ctx._cancel_called:
|
||||
# XXX: this should NEVER happen!
|
||||
# from ._debug import breakpoint
|
||||
# await breakpoint()
|
||||
# context was cancelled at the far end but was
|
||||
# not part of this end requesting that cancel
|
||||
# so raise for the local task to respond and handle.
|
||||
raise
|
||||
|
||||
# if the context was cancelled by client code
|
||||
|
|
@ -503,17 +468,17 @@ class Portal:
|
|||
|
||||
) as err:
|
||||
etype = type(err)
|
||||
# the context cancels itself on any cancel
|
||||
# causing error.
|
||||
|
||||
# cancel ourselves on any error.
|
||||
log.cancel(
|
||||
'Context cancelled for task, sending cancel request..\n'
|
||||
f'task:{cid}\n'
|
||||
f'actor:{uid}'
|
||||
)
|
||||
try:
|
||||
|
||||
if ctx.chan.connected():
|
||||
log.cancel(
|
||||
'Context cancelled for task, sending cancel request..\n'
|
||||
f'task:{cid}\n'
|
||||
f'actor:{uid}'
|
||||
)
|
||||
await ctx.cancel()
|
||||
except trio.BrokenResourceError:
|
||||
else:
|
||||
log.warning(
|
||||
'IPC connection for context is broken?\n'
|
||||
f'task:{cid}\n'
|
||||
|
|
@ -522,7 +487,12 @@ class Portal:
|
|||
|
||||
raise
|
||||
|
||||
else:
|
||||
finally:
|
||||
# in the case where a runtime nursery (due to internal bug)
|
||||
# or a remote actor transmits an error we want to be
|
||||
# sure we get the error the underlying feeder mem chan.
|
||||
# if it's not raised here it *should* be raised from the
|
||||
# msg loop nursery right?
|
||||
if ctx.chan.connected():
|
||||
log.info(
|
||||
'Waiting on final context-task result for\n'
|
||||
|
|
@ -535,7 +505,6 @@ class Portal:
|
|||
f'value from callee `{result}`'
|
||||
)
|
||||
|
||||
finally:
|
||||
# though it should be impossible for any tasks
|
||||
# operating *in* this scope to have survived
|
||||
# we tear down the runtime feeder chan last
|
||||
|
|
|
|||
|
|
@ -254,9 +254,7 @@ async def open_root_actor(
|
|||
# tempn.start_soon(an.exited.wait)
|
||||
|
||||
logger.cancel("Shutting down root actor")
|
||||
await actor.cancel(
|
||||
requesting_uid=actor.uid,
|
||||
)
|
||||
await actor.cancel()
|
||||
finally:
|
||||
_state._current_actor = None
|
||||
|
||||
|
|
|
|||
|
|
@ -28,10 +28,9 @@ import inspect
|
|||
import signal
|
||||
import sys
|
||||
from typing import (
|
||||
Any,
|
||||
Any, Optional,
|
||||
Union, TYPE_CHECKING,
|
||||
Callable,
|
||||
Union,
|
||||
TYPE_CHECKING,
|
||||
)
|
||||
import uuid
|
||||
from types import ModuleType
|
||||
|
|
@ -45,10 +44,7 @@ import trio # type: ignore
|
|||
from trio_typing import TaskStatus
|
||||
|
||||
from ._ipc import Channel
|
||||
from ._context import (
|
||||
mk_context,
|
||||
Context,
|
||||
)
|
||||
from ._streaming import Context
|
||||
from .log import get_logger
|
||||
from ._exceptions import (
|
||||
pack_error,
|
||||
|
|
@ -57,6 +53,7 @@ from ._exceptions import (
|
|||
is_multi_cancelled,
|
||||
ContextCancelled,
|
||||
TransportClosed,
|
||||
StreamOverrun,
|
||||
)
|
||||
from . import _debug
|
||||
from ._discovery import get_arbiter
|
||||
|
|
@ -82,7 +79,7 @@ async def _invoke(
|
|||
|
||||
is_rpc: bool = True,
|
||||
task_status: TaskStatus[
|
||||
Union[Context, BaseException]
|
||||
Union[trio.CancelScope, BaseException]
|
||||
] = trio.TASK_STATUS_IGNORED,
|
||||
):
|
||||
'''
|
||||
|
|
@ -100,16 +97,9 @@ async def _invoke(
|
|||
|
||||
cancel_scope = trio.CancelScope()
|
||||
# activated cancel scope ref
|
||||
cs: trio.CancelScope | None = None
|
||||
cs: Optional[trio.CancelScope] = None
|
||||
|
||||
ctx = actor.get_context(
|
||||
chan,
|
||||
cid,
|
||||
# We shouldn't ever need to pass this through right?
|
||||
# it's up to the soon-to-be called rpc task to
|
||||
# open the stream with this option.
|
||||
# allow_overruns=True,
|
||||
)
|
||||
ctx = actor.get_context(chan, cid)
|
||||
context: bool = False
|
||||
|
||||
if getattr(func, '_tractor_stream_function', False):
|
||||
|
|
@ -148,10 +138,7 @@ async def _invoke(
|
|||
):
|
||||
raise TypeError(f'{func} must be an async function!')
|
||||
|
||||
try:
|
||||
coro = func(**kwargs)
|
||||
except TypeError:
|
||||
raise
|
||||
coro = func(**kwargs)
|
||||
|
||||
if inspect.isasyncgen(coro):
|
||||
await chan.send({'functype': 'asyncgen', 'cid': cid})
|
||||
|
|
@ -163,8 +150,7 @@ async def _invoke(
|
|||
# of the async gen in order to be sure the cancel
|
||||
# is propagated!
|
||||
with cancel_scope as cs:
|
||||
ctx._scope = cs
|
||||
task_status.started(ctx)
|
||||
task_status.started(cs)
|
||||
async with aclosing(coro) as agen:
|
||||
async for item in agen:
|
||||
# TODO: can we send values back in here?
|
||||
|
|
@ -190,8 +176,7 @@ async def _invoke(
|
|||
# manualy construct the response dict-packet-responses as
|
||||
# above
|
||||
with cancel_scope as cs:
|
||||
ctx._scope = cs
|
||||
task_status.started(ctx)
|
||||
task_status.started(cs)
|
||||
await coro
|
||||
|
||||
if not cs.cancelled_caught:
|
||||
|
|
@ -204,26 +189,19 @@ async def _invoke(
|
|||
await chan.send({'functype': 'context', 'cid': cid})
|
||||
|
||||
try:
|
||||
async with trio.open_nursery() as nurse:
|
||||
ctx._scope_nursery = nurse
|
||||
ctx._scope = nurse.cancel_scope
|
||||
task_status.started(ctx)
|
||||
async with trio.open_nursery() as scope_nursery:
|
||||
ctx._scope_nursery = scope_nursery
|
||||
cs = scope_nursery.cancel_scope
|
||||
task_status.started(cs)
|
||||
res = await coro
|
||||
await chan.send({'return': res, 'cid': cid})
|
||||
|
||||
# XXX: do we ever trigger this block any more?
|
||||
except (
|
||||
BaseExceptionGroup,
|
||||
trio.Cancelled,
|
||||
):
|
||||
except BaseExceptionGroup:
|
||||
# if a context error was set then likely
|
||||
# thei multierror was raised due to that
|
||||
if ctx._remote_error is not None:
|
||||
raise ctx._remote_error
|
||||
if ctx._error is not None:
|
||||
raise ctx._error from None
|
||||
|
||||
# maybe TODO: pack in ``trio.Cancelled.__traceback__`` here
|
||||
# so they can be unwrapped and displayed on the caller
|
||||
# side?
|
||||
raise
|
||||
|
||||
finally:
|
||||
|
|
@ -235,67 +213,38 @@ async def _invoke(
|
|||
# associated child isn't in debug any more
|
||||
await _debug.maybe_wait_for_debugger()
|
||||
ctx = actor._contexts.pop((chan.uid, cid))
|
||||
|
||||
if ctx:
|
||||
log.runtime(
|
||||
f'Context entrypoint {func} was terminated:\n{ctx}'
|
||||
)
|
||||
|
||||
if ctx.cancelled_caught:
|
||||
assert cs
|
||||
if cs.cancelled_caught:
|
||||
|
||||
# first check for and raise any remote error
|
||||
# before raising any context cancelled case
|
||||
# so that real remote errors don't get masked as
|
||||
# ``ContextCancelled``s.
|
||||
re = ctx._remote_error
|
||||
if re:
|
||||
ctx._maybe_raise_remote_err(re)
|
||||
# TODO: pack in ``trio.Cancelled.__traceback__`` here
|
||||
# so they can be unwrapped and displayed on the caller
|
||||
# side!
|
||||
|
||||
fname = func.__name__
|
||||
cs: trio.CancelScope = ctx._scope
|
||||
if cs.cancel_called:
|
||||
canceller = ctx._cancelled_remote
|
||||
# await _debug.breakpoint()
|
||||
if ctx._cancel_called:
|
||||
msg = f'`{fname}()` cancelled itself'
|
||||
|
||||
# NOTE / TODO: if we end up having
|
||||
# ``Actor._cancel_task()`` call
|
||||
# ``Context.cancel()`` directly, we're going to
|
||||
# need to change this logic branch since it will
|
||||
# always enter..
|
||||
if ctx._cancel_called:
|
||||
msg = f'`{fname}()`@{actor.uid} cancelled itself'
|
||||
|
||||
else:
|
||||
msg = (
|
||||
f'`{fname}()`@{actor.uid} '
|
||||
'was remotely cancelled by '
|
||||
)
|
||||
|
||||
# if the channel which spawned the ctx is the
|
||||
# one that cancelled it then we report that, vs.
|
||||
# it being some other random actor that for ex.
|
||||
# some actor who calls `Portal.cancel_actor()`
|
||||
# and by side-effect cancels this ctx.
|
||||
if canceller == ctx.chan.uid:
|
||||
msg += f'its caller {canceller}'
|
||||
else:
|
||||
msg += f'remote actor {canceller}'
|
||||
|
||||
# TODO: does this ever get set any more or can
|
||||
# we remove it?
|
||||
if ctx._cancel_msg:
|
||||
msg += f' with msg:\n{ctx._cancel_msg}'
|
||||
|
||||
# task-contex was either cancelled by request using
|
||||
# ``Portal.cancel_actor()`` or ``Context.cancel()``
|
||||
# on the far end, or it was cancelled by the local
|
||||
# (callee) task, so relay this cancel signal to the
|
||||
# other side.
|
||||
raise ContextCancelled(
|
||||
msg,
|
||||
suberror_type=trio.Cancelled,
|
||||
canceller=canceller,
|
||||
elif cs.cancel_called:
|
||||
msg = (
|
||||
f'`{fname}()` was remotely cancelled by its caller '
|
||||
f'{ctx.chan.uid}'
|
||||
)
|
||||
|
||||
if ctx._cancel_msg:
|
||||
msg += f' with msg:\n{ctx._cancel_msg}'
|
||||
|
||||
# task-contex was cancelled so relay to the cancel to caller
|
||||
raise ContextCancelled(
|
||||
msg,
|
||||
suberror_type=trio.Cancelled,
|
||||
)
|
||||
|
||||
else:
|
||||
# regular async function
|
||||
try:
|
||||
|
|
@ -310,17 +259,12 @@ async def _invoke(
|
|||
)
|
||||
|
||||
with cancel_scope as cs:
|
||||
ctx._scope = cs
|
||||
task_status.started(ctx)
|
||||
task_status.started(cs)
|
||||
result = await coro
|
||||
fname = func.__name__
|
||||
log.runtime(f'{fname}() result: {result}')
|
||||
log.cancel(f'result: {result}')
|
||||
if not failed_resp:
|
||||
# only send result if we know IPC isn't down
|
||||
await chan.send(
|
||||
{'return': result,
|
||||
'cid': cid}
|
||||
)
|
||||
await chan.send({'return': result, 'cid': cid})
|
||||
|
||||
except (
|
||||
Exception,
|
||||
|
|
@ -363,7 +307,6 @@ async def _invoke(
|
|||
# always ship errors back to caller
|
||||
err_msg = pack_error(err, tb=tb)
|
||||
err_msg['cid'] = cid
|
||||
|
||||
try:
|
||||
await chan.send(err_msg)
|
||||
|
||||
|
|
@ -380,21 +323,14 @@ async def _invoke(
|
|||
f"Failed to ship error to caller @ {chan.uid} !?"
|
||||
)
|
||||
|
||||
# error is probably from above coro running code *not from the
|
||||
# underlyingn rpc invocation* since a scope was never allocated
|
||||
# around actual coroutine await.
|
||||
if ctx._scope is None:
|
||||
# we don't ever raise directly here to allow the
|
||||
# msg-loop-scheduler to continue running for this
|
||||
# channel.
|
||||
if cs is None:
|
||||
# error is from above code not from rpc invocation
|
||||
task_status.started(err)
|
||||
|
||||
finally:
|
||||
# RPC task bookeeping
|
||||
try:
|
||||
ctx, func, is_complete = actor._rpc_tasks.pop(
|
||||
(chan, cid)
|
||||
)
|
||||
scope, func, is_complete = actor._rpc_tasks.pop((chan, cid))
|
||||
is_complete.set()
|
||||
|
||||
except KeyError:
|
||||
|
|
@ -403,9 +339,6 @@ async def _invoke(
|
|||
# cancel scope will not have been inserted yet
|
||||
log.warning(
|
||||
f"Task {func} likely errored or cancelled before start")
|
||||
else:
|
||||
log.cancel(f'{func.__name__}({kwargs}) failed?')
|
||||
|
||||
finally:
|
||||
if not actor._rpc_tasks:
|
||||
log.runtime("All RPC tasks have completed")
|
||||
|
|
@ -467,20 +400,16 @@ class Actor:
|
|||
msg_buffer_size: int = 2**6
|
||||
|
||||
# nursery placeholders filled in by `async_main()` after fork
|
||||
_root_n: trio.Nursery | None = None
|
||||
_service_n: trio.Nursery | None = None
|
||||
_server_n: trio.Nursery | None = None
|
||||
_root_n: Optional[trio.Nursery] = None
|
||||
_service_n: Optional[trio.Nursery] = None
|
||||
_server_n: Optional[trio.Nursery] = None
|
||||
|
||||
# Information about `__main__` from parent
|
||||
_parent_main_data: dict[str, str]
|
||||
_parent_chan_cs: trio.CancelScope | None = None
|
||||
_parent_chan_cs: Optional[trio.CancelScope] = None
|
||||
|
||||
# syncs for setup/teardown sequences
|
||||
_server_down: trio.Event | None = None
|
||||
|
||||
# user toggled crash handling (including monkey-patched in
|
||||
# `trio.open_nursery()` via `.trionics._supervisor` B)
|
||||
_debug_mode: bool = False
|
||||
_server_down: Optional[trio.Event] = None
|
||||
|
||||
# if started on ``asycio`` running ``trio`` in guest mode
|
||||
_infected_aio: bool = False
|
||||
|
|
@ -496,8 +425,8 @@ class Actor:
|
|||
enable_modules: list[str] = [],
|
||||
uid: str | None = None,
|
||||
loglevel: str | None = None,
|
||||
arbiter_addr: tuple[str, int] | None = None,
|
||||
spawn_method: str | None = None
|
||||
arbiter_addr: Optional[tuple[str, int]] = None,
|
||||
spawn_method: Optional[str] = None
|
||||
) -> None:
|
||||
'''
|
||||
This constructor is called in the parent actor **before** the spawning
|
||||
|
|
@ -508,7 +437,6 @@ class Actor:
|
|||
self.uid = (name, uid or str(uuid.uuid4()))
|
||||
|
||||
self._cancel_complete = trio.Event()
|
||||
self._cancel_called_by_remote: tuple[str, tuple] | None = None
|
||||
self._cancel_called: bool = False
|
||||
|
||||
# retreive and store parent `__main__` data which
|
||||
|
|
@ -547,7 +475,7 @@ class Actor:
|
|||
# (chan, cid) -> (cancel_scope, func)
|
||||
self._rpc_tasks: dict[
|
||||
tuple[Channel, str],
|
||||
tuple[Context, Callable, trio.Event]
|
||||
tuple[trio.CancelScope, Callable, trio.Event]
|
||||
] = {}
|
||||
|
||||
# map {actor uids -> Context}
|
||||
|
|
@ -557,8 +485,9 @@ class Actor:
|
|||
] = {}
|
||||
|
||||
self._listeners: list[trio.abc.Listener] = []
|
||||
self._parent_chan: Channel | None = None
|
||||
self._forkserver_info: tuple | None = None
|
||||
self._parent_chan: Optional[Channel] = None
|
||||
self._forkserver_info: Optional[
|
||||
tuple[Any, Any, Any, Any, Any]] = None
|
||||
self._actoruid2nursery: dict[
|
||||
tuple[str, str],
|
||||
ActorNursery | None,
|
||||
|
|
@ -649,7 +578,7 @@ class Actor:
|
|||
self._no_more_peers = trio.Event() # unset
|
||||
|
||||
chan = Channel.from_stream(stream)
|
||||
uid: tuple[str, str] | None = chan.uid
|
||||
uid: Optional[tuple[str, str]] = chan.uid
|
||||
log.runtime(f"New connection to us {chan}")
|
||||
|
||||
# send/receive initial handshake response
|
||||
|
|
@ -697,7 +626,7 @@ class Actor:
|
|||
# append new channel
|
||||
self._peers[uid].append(chan)
|
||||
|
||||
local_nursery: ActorNursery | None = None # noqa
|
||||
local_nursery: Optional[ActorNursery] = None # noqa
|
||||
disconnected: bool = False
|
||||
|
||||
# Begin channel management - respond to remote requests and
|
||||
|
|
@ -721,8 +650,8 @@ class Actor:
|
|||
if (
|
||||
local_nursery
|
||||
):
|
||||
if chan._cancel_called:
|
||||
log.cancel(f"Waiting on cancel request to peer {chan.uid}")
|
||||
|
||||
log.cancel(f"Waiting on cancel request to peer {chan.uid}")
|
||||
# XXX: this is a soft wait on the channel (and its
|
||||
# underlying transport protocol) to close from the
|
||||
# remote peer side since we presume that any channel
|
||||
|
|
@ -855,15 +784,75 @@ class Actor:
|
|||
f'\n{msg}')
|
||||
return
|
||||
|
||||
return await ctx._deliver_msg(msg)
|
||||
send_chan = ctx._send_chan
|
||||
|
||||
log.runtime(f"Delivering {msg} from {chan.uid} to caller {cid}")
|
||||
|
||||
# XXX: we do **not** maintain backpressure and instead
|
||||
# opt to relay stream overrun errors to the sender.
|
||||
try:
|
||||
send_chan.send_nowait(msg)
|
||||
# if an error is deteced we should always
|
||||
# expect it to be raised by any context (stream)
|
||||
# consumer task
|
||||
await ctx._maybe_raise_from_remote_msg(msg)
|
||||
|
||||
except trio.BrokenResourceError:
|
||||
# TODO: what is the right way to handle the case where the
|
||||
# local task has already sent a 'stop' / StopAsyncInteration
|
||||
# to the other side but and possibly has closed the local
|
||||
# feeder mem chan? Do we wait for some kind of ack or just
|
||||
# let this fail silently and bubble up (currently)?
|
||||
|
||||
# XXX: local consumer has closed their side
|
||||
# so cancel the far end streaming task
|
||||
log.warning(f"{send_chan} consumer is already closed")
|
||||
return
|
||||
|
||||
except trio.WouldBlock:
|
||||
# XXX: always push an error even if the local
|
||||
# receiver is in overrun state.
|
||||
await ctx._maybe_raise_from_remote_msg(msg)
|
||||
|
||||
uid = chan.uid
|
||||
lines = [
|
||||
'Task context stream was overrun',
|
||||
f'local task: {cid} @ {self.uid}',
|
||||
f'remote sender: {uid}',
|
||||
]
|
||||
if not ctx._stream_opened:
|
||||
lines.insert(
|
||||
1,
|
||||
f'\n*** No stream open on `{self.uid[0]}` side! ***\n'
|
||||
)
|
||||
text = '\n'.join(lines)
|
||||
|
||||
if ctx._backpressure:
|
||||
log.warning(text)
|
||||
try:
|
||||
await send_chan.send(msg)
|
||||
except trio.BrokenResourceError:
|
||||
# XXX: local consumer has closed their side
|
||||
# so cancel the far end streaming task
|
||||
log.warning(f"{chan} is already closed")
|
||||
else:
|
||||
try:
|
||||
raise StreamOverrun(text) from None
|
||||
except StreamOverrun as err:
|
||||
err_msg = pack_error(err)
|
||||
err_msg['cid'] = cid
|
||||
try:
|
||||
await chan.send(err_msg)
|
||||
except trio.BrokenResourceError:
|
||||
# XXX: local consumer has closed their side
|
||||
# so cancel the far end streaming task
|
||||
log.warning(f"{chan} is already closed")
|
||||
|
||||
def get_context(
|
||||
self,
|
||||
chan: Channel,
|
||||
cid: str,
|
||||
|
||||
msg_buffer_size: int | None = None,
|
||||
allow_overruns: bool = False,
|
||||
msg_buffer_size: Optional[int] = None,
|
||||
|
||||
) -> Context:
|
||||
'''
|
||||
|
|
@ -879,7 +868,6 @@ class Actor:
|
|||
assert actor_uid
|
||||
try:
|
||||
ctx = self._contexts[(actor_uid, cid)]
|
||||
ctx._allow_overruns = allow_overruns
|
||||
|
||||
# adjust buffer size if specified
|
||||
state = ctx._send_chan._state # type: ignore
|
||||
|
|
@ -887,11 +875,15 @@ class Actor:
|
|||
state.max_buffer_size = msg_buffer_size
|
||||
|
||||
except KeyError:
|
||||
ctx = mk_context(
|
||||
send_chan: trio.MemorySendChannel
|
||||
recv_chan: trio.MemoryReceiveChannel
|
||||
send_chan, recv_chan = trio.open_memory_channel(
|
||||
msg_buffer_size or self.msg_buffer_size)
|
||||
ctx = Context(
|
||||
chan,
|
||||
cid,
|
||||
msg_buffer_size=msg_buffer_size or self.msg_buffer_size,
|
||||
_allow_overruns=allow_overruns,
|
||||
_send_chan=send_chan,
|
||||
_recv_chan=recv_chan,
|
||||
)
|
||||
self._contexts[(actor_uid, cid)] = ctx
|
||||
|
||||
|
|
@ -903,8 +895,7 @@ class Actor:
|
|||
ns: str,
|
||||
func: str,
|
||||
kwargs: dict,
|
||||
msg_buffer_size: int | None = None,
|
||||
allow_overruns: bool = False,
|
||||
msg_buffer_size: Optional[int] = None,
|
||||
|
||||
) -> Context:
|
||||
'''
|
||||
|
|
@ -918,16 +909,9 @@ class Actor:
|
|||
'''
|
||||
cid = str(uuid.uuid4())
|
||||
assert chan.uid
|
||||
ctx = self.get_context(
|
||||
chan,
|
||||
cid,
|
||||
msg_buffer_size=msg_buffer_size,
|
||||
allow_overruns=allow_overruns,
|
||||
)
|
||||
ctx = self.get_context(chan, cid, msg_buffer_size=msg_buffer_size)
|
||||
log.runtime(f"Sending cmd to {chan.uid}: {ns}.{func}({kwargs})")
|
||||
await chan.send(
|
||||
{'cmd': (ns, func, kwargs, self.uid, cid)}
|
||||
)
|
||||
await chan.send({'cmd': (ns, func, kwargs, self.uid, cid)})
|
||||
|
||||
# Wait on first response msg and validate; this should be
|
||||
# immediate.
|
||||
|
|
@ -937,11 +921,7 @@ class Actor:
|
|||
if 'error' in first_msg:
|
||||
raise unpack_error(first_msg, chan)
|
||||
|
||||
elif functype not in (
|
||||
'asyncfunc',
|
||||
'asyncgen',
|
||||
'context',
|
||||
):
|
||||
elif functype not in ('asyncfunc', 'asyncgen', 'context'):
|
||||
raise ValueError(f"{first_msg} is an invalid response packet?")
|
||||
|
||||
ctx._remote_func_type = functype
|
||||
|
|
@ -949,8 +929,8 @@ class Actor:
|
|||
|
||||
async def _from_parent(
|
||||
self,
|
||||
parent_addr: tuple[str, int] | None,
|
||||
) -> tuple[Channel, tuple[str, int] | None]:
|
||||
parent_addr: Optional[tuple[str, int]],
|
||||
) -> tuple[Channel, Optional[tuple[str, int]]]:
|
||||
try:
|
||||
# Connect back to the parent actor and conduct initial
|
||||
# handshake. From this point on if we error, we
|
||||
|
|
@ -963,7 +943,7 @@ class Actor:
|
|||
# Initial handshake: swap names.
|
||||
await self._do_handshake(chan)
|
||||
|
||||
accept_addr: tuple[str, int] | None = None
|
||||
accept_addr: Optional[tuple[str, int]] = None
|
||||
|
||||
if self._spawn_method == "trio":
|
||||
# Receive runtime state from our parent
|
||||
|
|
@ -1000,7 +980,7 @@ class Actor:
|
|||
log.warning(
|
||||
f"Failed to connect to parent @ {parent_addr},"
|
||||
" closing server")
|
||||
await self.cancel(requesting_uid=self.uid)
|
||||
await self.cancel()
|
||||
raise
|
||||
|
||||
async def _serve_forever(
|
||||
|
|
@ -1022,7 +1002,7 @@ class Actor:
|
|||
self._server_down = trio.Event()
|
||||
try:
|
||||
async with trio.open_nursery() as server_n:
|
||||
listeners: list[trio.abc.Listener] = await server_n.start(
|
||||
l: list[trio.abc.Listener] = await server_n.start(
|
||||
partial(
|
||||
trio.serve_tcp,
|
||||
self._stream_handler,
|
||||
|
|
@ -1033,13 +1013,10 @@ class Actor:
|
|||
host=accept_host,
|
||||
)
|
||||
)
|
||||
sockets: list[trio.socket] = [
|
||||
getattr(listener, 'socket', 'unknown socket')
|
||||
for listener in listeners
|
||||
]
|
||||
log.runtime(
|
||||
f'Started tcp server(s) on {sockets}')
|
||||
self._listeners.extend(listeners)
|
||||
"Started tcp server(s) on"
|
||||
f" {[getattr(l, 'socket', 'unknown socket') for l in l]}")
|
||||
self._listeners.extend(l)
|
||||
task_status.started(server_n)
|
||||
finally:
|
||||
# signal the server is down since nursery above terminated
|
||||
|
|
@ -1056,11 +1033,7 @@ class Actor:
|
|||
assert self._service_n
|
||||
self._service_n.start_soon(self.cancel)
|
||||
|
||||
async def cancel(
|
||||
self,
|
||||
requesting_uid: tuple[str, str],
|
||||
|
||||
) -> bool:
|
||||
async def cancel(self) -> bool:
|
||||
'''
|
||||
Cancel this actor's runtime.
|
||||
|
||||
|
|
@ -1074,7 +1047,6 @@ class Actor:
|
|||
|
||||
'''
|
||||
log.cancel(f"{self.uid} is trying to cancel")
|
||||
self._cancel_called_by_remote: tuple = requesting_uid
|
||||
self._cancel_called = True
|
||||
|
||||
# cancel all ongoing rpc tasks
|
||||
|
|
@ -1088,7 +1060,7 @@ class Actor:
|
|||
dbcs.cancel()
|
||||
|
||||
# kill all ongoing tasks
|
||||
await self.cancel_rpc_tasks(requesting_uid=requesting_uid)
|
||||
await self.cancel_rpc_tasks()
|
||||
|
||||
# stop channel server
|
||||
self.cancel_server()
|
||||
|
|
@ -1114,13 +1086,7 @@ class Actor:
|
|||
# for n in root.child_nurseries:
|
||||
# n.cancel_scope.cancel()
|
||||
|
||||
async def _cancel_task(
|
||||
self,
|
||||
cid: str,
|
||||
chan: Channel,
|
||||
|
||||
requesting_uid: tuple[str, str] | None = None,
|
||||
) -> bool:
|
||||
async def _cancel_task(self, cid, chan):
|
||||
'''
|
||||
Cancel a local task by call-id / channel.
|
||||
|
||||
|
|
@ -1135,51 +1101,35 @@ class Actor:
|
|||
try:
|
||||
# this ctx based lookup ensures the requested task to
|
||||
# be cancelled was indeed spawned by a request from this channel
|
||||
ctx, func, is_complete = self._rpc_tasks[(chan, cid)]
|
||||
scope = ctx._scope
|
||||
scope, func, is_complete = self._rpc_tasks[(chan, cid)]
|
||||
except KeyError:
|
||||
log.cancel(f"{cid} has already completed/terminated?")
|
||||
return True
|
||||
return
|
||||
|
||||
log.cancel(
|
||||
f"Cancelling task:\ncid: {cid}\nfunc: {func}\n"
|
||||
f"peer: {chan.uid}\n")
|
||||
|
||||
if (
|
||||
ctx._cancelled_remote is None
|
||||
and requesting_uid
|
||||
):
|
||||
ctx._cancelled_remote: tuple = requesting_uid
|
||||
|
||||
# don't allow cancelling this function mid-execution
|
||||
# (is this necessary?)
|
||||
if func is self._cancel_task:
|
||||
return True
|
||||
return
|
||||
|
||||
# TODO: shouldn't we eventually be calling ``Context.cancel()``
|
||||
# directly here instead (since that method can handle both
|
||||
# side's calls into it?
|
||||
scope.cancel()
|
||||
|
||||
# wait for _invoke to mark the task complete
|
||||
log.runtime(
|
||||
'Waiting on task to cancel:\n'
|
||||
f'cid: {cid}\nfunc: {func}\n'
|
||||
f'peer: {chan.uid}\n'
|
||||
)
|
||||
f"Waiting on task to cancel:\ncid: {cid}\nfunc: {func}\n"
|
||||
f"peer: {chan.uid}\n")
|
||||
await is_complete.wait()
|
||||
|
||||
log.runtime(
|
||||
f"Sucessfully cancelled task:\ncid: {cid}\nfunc: {func}\n"
|
||||
f"peer: {chan.uid}\n")
|
||||
|
||||
return True
|
||||
|
||||
async def cancel_rpc_tasks(
|
||||
self,
|
||||
only_chan: Channel | None = None,
|
||||
requesting_uid: tuple[str, str] | None = None,
|
||||
|
||||
only_chan: Optional[Channel] = None,
|
||||
) -> None:
|
||||
'''
|
||||
Cancel all existing RPC responder tasks using the cancel scope
|
||||
|
|
@ -1191,7 +1141,7 @@ class Actor:
|
|||
log.cancel(f"Cancelling all {len(tasks)} rpc tasks:\n{tasks} ")
|
||||
for (
|
||||
(chan, cid),
|
||||
(ctx, func, is_complete),
|
||||
(scope, func, is_complete),
|
||||
) in tasks.copy().items():
|
||||
if only_chan is not None:
|
||||
if only_chan != chan:
|
||||
|
|
@ -1199,11 +1149,7 @@ class Actor:
|
|||
|
||||
# TODO: this should really done in a nursery batch
|
||||
if func != self._cancel_task:
|
||||
await self._cancel_task(
|
||||
cid,
|
||||
chan,
|
||||
requesting_uid=requesting_uid,
|
||||
)
|
||||
await self._cancel_task(cid, chan)
|
||||
|
||||
log.cancel(
|
||||
f"Waiting for remaining rpc tasks to complete {tasks}")
|
||||
|
|
@ -1220,7 +1166,7 @@ class Actor:
|
|||
self._server_n.cancel_scope.cancel()
|
||||
|
||||
@property
|
||||
def accept_addr(self) -> tuple[str, int] | None:
|
||||
def accept_addr(self) -> Optional[tuple[str, int]]:
|
||||
'''
|
||||
Primary address to which the channel server is bound.
|
||||
|
||||
|
|
@ -1272,7 +1218,7 @@ class Actor:
|
|||
|
||||
async def async_main(
|
||||
actor: Actor,
|
||||
accept_addr: tuple[str, int] | None = None,
|
||||
accept_addr: Optional[tuple[str, int]] = None,
|
||||
|
||||
# XXX: currently ``parent_addr`` is only needed for the
|
||||
# ``multiprocessing`` backend (which pickles state sent to
|
||||
|
|
@ -1281,7 +1227,7 @@ async def async_main(
|
|||
# change this to a simple ``is_subactor: bool`` which will
|
||||
# be False when running as root actor and True when as
|
||||
# a subactor.
|
||||
parent_addr: tuple[str, int] | None = None,
|
||||
parent_addr: Optional[tuple[str, int]] = None,
|
||||
task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
|
||||
|
||||
) -> None:
|
||||
|
|
@ -1289,8 +1235,8 @@ async def async_main(
|
|||
Actor runtime entrypoint; start the IPC channel server, maybe connect
|
||||
back to the parent, and startup all core machinery tasks.
|
||||
|
||||
A "root" (or "top-level") nursery for this actor is opened here and
|
||||
when cancelled/terminated effectively closes the actor's "runtime".
|
||||
A "root-most" (or "top-level") nursery for this actor is opened here
|
||||
and when cancelled effectively cancels the actor.
|
||||
|
||||
'''
|
||||
# attempt to retreive ``trio``'s sigint handler and stash it
|
||||
|
|
@ -1384,15 +1330,13 @@ async def async_main(
|
|||
)
|
||||
)
|
||||
log.runtime("Waiting on service nursery to complete")
|
||||
log.runtime(
|
||||
"Service nursery complete\n"
|
||||
"Waiting on root nursery to complete"
|
||||
)
|
||||
log.runtime("Service nursery complete")
|
||||
log.runtime("Waiting on root nursery to complete")
|
||||
|
||||
# Blocks here as expected until the root nursery is
|
||||
# killed (i.e. this actor is cancelled or signalled by the parent)
|
||||
except Exception as err:
|
||||
log.runtime("Closing all actor lifetime contexts")
|
||||
log.info("Closing all actor lifetime contexts")
|
||||
actor.lifetime_stack.close()
|
||||
|
||||
if not registered_with_arbiter:
|
||||
|
|
@ -1413,14 +1357,7 @@ async def async_main(
|
|||
await try_ship_error_to_parent(actor._parent_chan, err)
|
||||
|
||||
# always!
|
||||
match err:
|
||||
case ContextCancelled():
|
||||
log.cancel(
|
||||
f'Actor: {actor.uid} was task-context-cancelled with,\n'
|
||||
f'str(err)'
|
||||
)
|
||||
case _:
|
||||
log.exception("Actor errored:")
|
||||
log.exception("Actor errored:")
|
||||
raise
|
||||
|
||||
finally:
|
||||
|
|
@ -1487,16 +1424,15 @@ async def process_messages(
|
|||
|
||||
) -> bool:
|
||||
'''
|
||||
This is the per-channel, low level RPC task scheduler loop.
|
||||
Process messages for the IPC transport channel async-RPC style.
|
||||
|
||||
Receive multiplexed RPC request messages from some remote process,
|
||||
spawn handler tasks depending on request type and deliver responses
|
||||
or boxed errors back to the remote caller (task).
|
||||
Receive multiplexed RPC requests, spawn handler tasks and deliver
|
||||
responses over or boxed errors back to the "caller" task.
|
||||
|
||||
'''
|
||||
# TODO: once https://github.com/python-trio/trio/issues/467 gets
|
||||
# worked out we'll likely want to use that!
|
||||
msg: dict | None = None
|
||||
msg = None
|
||||
nursery_cancelled_before_task: bool = False
|
||||
|
||||
log.runtime(f"Entering msg loop for {chan} from {chan.uid}")
|
||||
|
|
@ -1518,10 +1454,7 @@ async def process_messages(
|
|||
|
||||
for (channel, cid) in actor._rpc_tasks.copy():
|
||||
if channel is chan:
|
||||
await actor._cancel_task(
|
||||
cid,
|
||||
channel,
|
||||
)
|
||||
await actor._cancel_task(cid, channel)
|
||||
|
||||
log.runtime(
|
||||
f"Msg loop signalled to terminate for"
|
||||
|
|
@ -1535,14 +1468,12 @@ async def process_messages(
|
|||
cid = msg.get('cid')
|
||||
if cid:
|
||||
# deliver response to local caller/waiter
|
||||
# via its per-remote-context memory channel.
|
||||
await actor._push_result(chan, cid, msg)
|
||||
|
||||
log.runtime(
|
||||
f"Waiting on next msg for {chan} from {chan.uid}")
|
||||
continue
|
||||
|
||||
# TODO: implement with ``match:`` syntax?
|
||||
# process command request
|
||||
try:
|
||||
ns, funcname, kwargs, actorid, cid = msg['cmd']
|
||||
|
|
@ -1562,12 +1493,13 @@ async def process_messages(
|
|||
f"{ns}.{funcname}({kwargs})")
|
||||
|
||||
if ns == 'self':
|
||||
if funcname == 'cancel':
|
||||
func = actor.cancel
|
||||
kwargs['requesting_uid'] = chan.uid
|
||||
func = getattr(actor, funcname)
|
||||
|
||||
# don't start entire actor runtime cancellation
|
||||
# if this actor is currently in debug mode!
|
||||
if funcname == 'cancel':
|
||||
|
||||
# don't start entire actor runtime
|
||||
# cancellation if this actor is in debug
|
||||
# mode
|
||||
pdb_complete = _debug.Lock.local_pdb_complete
|
||||
if pdb_complete:
|
||||
await pdb_complete.wait()
|
||||
|
|
@ -1579,56 +1511,43 @@ async def process_messages(
|
|||
# msg loop and break out into
|
||||
# ``async_main()``
|
||||
log.cancel(
|
||||
"Actor runtime for was remotely cancelled "
|
||||
f"Actor {actor.uid} was remotely cancelled "
|
||||
f"by {chan.uid}"
|
||||
)
|
||||
await _invoke(
|
||||
actor,
|
||||
cid,
|
||||
chan,
|
||||
func,
|
||||
kwargs,
|
||||
is_rpc=False,
|
||||
actor, cid, chan, func, kwargs, is_rpc=False
|
||||
)
|
||||
|
||||
log.cancel(
|
||||
f'Cancelling msg loop for {chan.uid}'
|
||||
)
|
||||
loop_cs.cancel()
|
||||
break
|
||||
|
||||
if funcname == '_cancel_task':
|
||||
func = actor._cancel_task
|
||||
|
||||
# we immediately start the runtime machinery
|
||||
# shutdown
|
||||
# with trio.CancelScope(shield=True):
|
||||
kwargs['chan'] = chan
|
||||
target_cid = kwargs['cid']
|
||||
kwargs['requesting_uid'] = chan.uid
|
||||
log.cancel(
|
||||
f'Remote request to cancel task\n'
|
||||
f'remote actor: {chan.uid}\n'
|
||||
f'task: {target_cid}'
|
||||
)
|
||||
try:
|
||||
await _invoke(
|
||||
actor,
|
||||
cid,
|
||||
chan,
|
||||
func,
|
||||
kwargs,
|
||||
is_rpc=False,
|
||||
with trio.CancelScope(shield=True):
|
||||
# actor.cancel() was called so kill this
|
||||
# msg loop and break out into
|
||||
# ``async_main()``
|
||||
kwargs['chan'] = chan
|
||||
log.cancel(
|
||||
f'Remote request to cancel task\n'
|
||||
f'remote actor: {chan.uid}\n'
|
||||
f'task: {cid}'
|
||||
)
|
||||
except BaseException:
|
||||
log.exception("failed to cancel task?")
|
||||
|
||||
continue
|
||||
else:
|
||||
# normally registry methods, eg.
|
||||
# ``.register_actor()`` etc.
|
||||
func = getattr(actor, funcname)
|
||||
try:
|
||||
await _invoke(
|
||||
actor,
|
||||
cid,
|
||||
chan,
|
||||
func,
|
||||
kwargs,
|
||||
is_rpc=False,
|
||||
)
|
||||
except BaseException:
|
||||
log.exception("failed to cancel task?")
|
||||
|
||||
continue
|
||||
else:
|
||||
# complain to client about restricted modules
|
||||
try:
|
||||
|
|
@ -1643,49 +1562,34 @@ async def process_messages(
|
|||
log.runtime(f"Spawning task for {func}")
|
||||
assert actor._service_n
|
||||
try:
|
||||
ctx: Context = await actor._service_n.start(
|
||||
partial(
|
||||
_invoke,
|
||||
actor,
|
||||
cid,
|
||||
chan,
|
||||
func,
|
||||
kwargs,
|
||||
),
|
||||
cs = await actor._service_n.start(
|
||||
partial(_invoke, actor, cid, chan, func, kwargs),
|
||||
name=funcname,
|
||||
)
|
||||
|
||||
except (
|
||||
RuntimeError,
|
||||
BaseExceptionGroup,
|
||||
):
|
||||
# avoid reporting a benign race condition
|
||||
# during actor runtime teardown.
|
||||
nursery_cancelled_before_task: bool = True
|
||||
nursery_cancelled_before_task = True
|
||||
break
|
||||
|
||||
# in the lone case where a ``Context`` is not
|
||||
# delivered, it's likely going to be a locally
|
||||
# scoped exception from ``_invoke()`` itself.
|
||||
if isinstance(ctx, Exception):
|
||||
# never allow cancelling cancel requests (results in
|
||||
# deadlock and other weird behaviour)
|
||||
# if func != actor.cancel:
|
||||
if isinstance(cs, Exception):
|
||||
log.warning(
|
||||
f"Task for RPC func {func} failed with"
|
||||
f"{ctx}"
|
||||
)
|
||||
continue
|
||||
|
||||
f"{cs}")
|
||||
else:
|
||||
# mark that we have ongoing rpc tasks
|
||||
actor._ongoing_rpc_tasks = trio.Event()
|
||||
log.runtime(f"RPC func is {func}")
|
||||
|
||||
# store cancel scope such that the rpc task can be
|
||||
# cancelled gracefully if requested
|
||||
actor._rpc_tasks[(chan, cid)] = (
|
||||
ctx,
|
||||
func,
|
||||
trio.Event(),
|
||||
)
|
||||
cs, func, trio.Event())
|
||||
|
||||
log.runtime(
|
||||
f"Waiting on next msg for {chan} from {chan.uid}")
|
||||
|
|
@ -1726,15 +1630,7 @@ async def process_messages(
|
|||
else:
|
||||
# ship any "internal" exception (i.e. one from internal
|
||||
# machinery not from an rpc task) to parent
|
||||
match err:
|
||||
case ContextCancelled():
|
||||
log.cancel(
|
||||
f'Actor: {actor.uid} was context-cancelled with,\n'
|
||||
f'str(err)'
|
||||
)
|
||||
case _:
|
||||
log.exception("Actor errored:")
|
||||
|
||||
log.exception("Actor errored:")
|
||||
if actor._parent_chan:
|
||||
await try_ship_error_to_parent(actor._parent_chan, err)
|
||||
|
||||
|
|
@ -1746,8 +1642,7 @@ async def process_messages(
|
|||
# msg debugging for when he machinery is brokey
|
||||
log.runtime(
|
||||
f"Exiting msg loop for {chan} from {chan.uid} "
|
||||
f"with last msg:\n{msg}"
|
||||
)
|
||||
f"with last msg:\n{msg}")
|
||||
|
||||
# transport **was not** disconnected
|
||||
return False
|
||||
|
|
|
|||
|
|
@ -19,7 +19,6 @@ Machinery for actor process spawning using multiple backends.
|
|||
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import multiprocessing as mp
|
||||
import sys
|
||||
import platform
|
||||
from typing import (
|
||||
|
|
@ -54,6 +53,7 @@ from ._exceptions import ActorFailure
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from ._supervise import ActorNursery
|
||||
import multiprocessing as mp
|
||||
ProcessType = TypeVar('ProcessType', mp.Process, trio.Process)
|
||||
|
||||
log = get_logger('tractor')
|
||||
|
|
@ -70,6 +70,7 @@ _spawn_method: SpawnMethodKey = 'trio'
|
|||
|
||||
if platform.system() == 'Windows':
|
||||
|
||||
import multiprocessing as mp
|
||||
_ctx = mp.get_context("spawn")
|
||||
|
||||
async def proc_waiter(proc: mp.Process) -> None:
|
||||
|
|
@ -456,7 +457,7 @@ async def trio_proc(
|
|||
|
||||
# cancel result waiter that may have been spawned in
|
||||
# tandem if not done already
|
||||
log.cancel(
|
||||
log.warning(
|
||||
"Cancelling existing result waiter task for "
|
||||
f"{subactor.uid}")
|
||||
nursery.cancel_scope.cancel()
|
||||
|
|
|
|||
|
|
@ -23,6 +23,11 @@ from typing import (
|
|||
Any,
|
||||
)
|
||||
|
||||
import trio
|
||||
|
||||
from ._exceptions import NoRuntime
|
||||
|
||||
|
||||
_current_actor: Optional['Actor'] = None # type: ignore # noqa
|
||||
_runtime_vars: dict[str, Any] = {
|
||||
'_debug_mode': False,
|
||||
|
|
@ -32,11 +37,8 @@ _runtime_vars: dict[str, Any] = {
|
|||
|
||||
|
||||
def current_actor(err_on_no_runtime: bool = True) -> 'Actor': # type: ignore # noqa
|
||||
'''
|
||||
Get the process-local actor instance.
|
||||
|
||||
'''
|
||||
from ._exceptions import NoRuntime
|
||||
"""Get the process-local actor instance.
|
||||
"""
|
||||
if _current_actor is None and err_on_no_runtime:
|
||||
raise NoRuntime("No local actor has been initialized yet")
|
||||
|
||||
|
|
@ -44,20 +46,16 @@ def current_actor(err_on_no_runtime: bool = True) -> 'Actor': # type: ignore #
|
|||
|
||||
|
||||
def is_main_process() -> bool:
|
||||
'''
|
||||
Bool determining if this actor is running in the top-most process.
|
||||
|
||||
'''
|
||||
"""Bool determining if this actor is running in the top-most process.
|
||||
"""
|
||||
import multiprocessing as mp
|
||||
return mp.current_process().name == 'MainProcess'
|
||||
|
||||
|
||||
def debug_mode() -> bool:
|
||||
'''
|
||||
Bool determining if "debug mode" is on which enables
|
||||
"""Bool determining if "debug mode" is on which enables
|
||||
remote subactor pdb entry on crashes.
|
||||
|
||||
'''
|
||||
"""
|
||||
return bool(_runtime_vars['_debug_mode'])
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -14,36 +14,31 @@
|
|||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
"""
|
||||
Message stream types and APIs.
|
||||
|
||||
The machinery and types behind ``Context.open_stream()``
|
||||
|
||||
'''
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import inspect
|
||||
from contextlib import asynccontextmanager as acm
|
||||
from contextlib import asynccontextmanager
|
||||
from dataclasses import dataclass
|
||||
from typing import (
|
||||
Any,
|
||||
Optional,
|
||||
Callable,
|
||||
AsyncIterator,
|
||||
TYPE_CHECKING,
|
||||
AsyncGenerator,
|
||||
AsyncIterator
|
||||
)
|
||||
|
||||
import warnings
|
||||
|
||||
import trio
|
||||
|
||||
from ._exceptions import (
|
||||
unpack_error,
|
||||
)
|
||||
from ._ipc import Channel
|
||||
from ._exceptions import unpack_error, ContextCancelled
|
||||
from ._state import current_actor
|
||||
from .log import get_logger
|
||||
from .trionics import (
|
||||
broadcast_receiver,
|
||||
BroadcastReceiver,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ._context import Context
|
||||
from .trionics import broadcast_receiver, BroadcastReceiver
|
||||
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
|
@ -75,9 +70,9 @@ class MsgStream(trio.abc.Channel):
|
|||
'''
|
||||
def __init__(
|
||||
self,
|
||||
ctx: Context, # typing: ignore # noqa
|
||||
ctx: 'Context', # typing: ignore # noqa
|
||||
rx_chan: trio.MemoryReceiveChannel,
|
||||
_broadcaster: BroadcastReceiver | None = None,
|
||||
_broadcaster: Optional[BroadcastReceiver] = None,
|
||||
|
||||
) -> None:
|
||||
self._ctx = ctx
|
||||
|
|
@ -280,7 +275,7 @@ class MsgStream(trio.abc.Channel):
|
|||
# still need to consume msgs that are "in transit" from the far
|
||||
# end (eg. for ``Context.result()``).
|
||||
|
||||
@acm
|
||||
@asynccontextmanager
|
||||
async def subscribe(
|
||||
self,
|
||||
|
||||
|
|
@ -340,8 +335,8 @@ class MsgStream(trio.abc.Channel):
|
|||
Send a message over this stream to the far end.
|
||||
|
||||
'''
|
||||
if self._ctx._remote_error:
|
||||
raise self._ctx._remote_error # from None
|
||||
if self._ctx._error:
|
||||
raise self._ctx._error # from None
|
||||
|
||||
if self._closed:
|
||||
raise trio.ClosedResourceError('This stream was already closed')
|
||||
|
|
@ -349,11 +344,371 @@ class MsgStream(trio.abc.Channel):
|
|||
await self._ctx.chan.send({'yield': data, 'cid': self._ctx.cid})
|
||||
|
||||
|
||||
def stream(func: Callable) -> Callable:
|
||||
@dataclass
|
||||
class Context:
|
||||
'''
|
||||
Mark an async function as a streaming routine with ``@stream``.
|
||||
An inter-actor, ``trio`` task communication context.
|
||||
|
||||
NB: This class should never be instatiated directly, it is delivered
|
||||
by either runtime machinery to a remotely started task or by entering
|
||||
``Portal.open_context()``.
|
||||
|
||||
Allows maintaining task or protocol specific state between
|
||||
2 communicating actor tasks. A unique context is created on the
|
||||
callee side/end for every request to a remote actor from a portal.
|
||||
|
||||
A context can be cancelled and (possibly eventually restarted) from
|
||||
either side of the underlying IPC channel, open task oriented
|
||||
message streams and acts as an IPC aware inter-actor-task cancel
|
||||
scope.
|
||||
|
||||
'''
|
||||
chan: Channel
|
||||
cid: str
|
||||
|
||||
# these are the "feeder" channels for delivering
|
||||
# message values to the local task from the runtime
|
||||
# msg processing loop.
|
||||
_recv_chan: trio.MemoryReceiveChannel
|
||||
_send_chan: trio.MemorySendChannel
|
||||
|
||||
_remote_func_type: Optional[str] = None
|
||||
|
||||
# only set on the caller side
|
||||
_portal: Optional['Portal'] = None # type: ignore # noqa
|
||||
_result: Optional[Any] = False
|
||||
_error: Optional[BaseException] = None
|
||||
|
||||
# status flags
|
||||
_cancel_called: bool = False
|
||||
_cancel_msg: Optional[str] = None
|
||||
_enter_debugger_on_cancel: bool = True
|
||||
_started_called: bool = False
|
||||
_started_received: bool = False
|
||||
_stream_opened: bool = False
|
||||
|
||||
# only set on the callee side
|
||||
_scope_nursery: Optional[trio.Nursery] = None
|
||||
|
||||
_backpressure: bool = False
|
||||
|
||||
async def send_yield(self, data: Any) -> None:
|
||||
|
||||
warnings.warn(
|
||||
"`Context.send_yield()` is now deprecated. "
|
||||
"Use ``MessageStream.send()``. ",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
await self.chan.send({'yield': data, 'cid': self.cid})
|
||||
|
||||
async def send_stop(self) -> None:
|
||||
await self.chan.send({'stop': True, 'cid': self.cid})
|
||||
|
||||
async def _maybe_raise_from_remote_msg(
|
||||
self,
|
||||
msg: dict[str, Any],
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
(Maybe) unpack and raise a msg error into the local scope
|
||||
nursery for this context.
|
||||
|
||||
Acts as a form of "relay" for a remote error raised
|
||||
in the corresponding remote callee task.
|
||||
|
||||
'''
|
||||
error = msg.get('error')
|
||||
if error:
|
||||
# If this is an error message from a context opened by
|
||||
# ``Portal.open_context()`` we want to interrupt any ongoing
|
||||
# (child) tasks within that context to be notified of the remote
|
||||
# error relayed here.
|
||||
#
|
||||
# The reason we may want to raise the remote error immediately
|
||||
# is that there is no guarantee the associated local task(s)
|
||||
# will attempt to read from any locally opened stream any time
|
||||
# soon.
|
||||
#
|
||||
# NOTE: this only applies when
|
||||
# ``Portal.open_context()`` has been called since it is assumed
|
||||
# (currently) that other portal APIs (``Portal.run()``,
|
||||
# ``.run_in_actor()``) do their own error checking at the point
|
||||
# of the call and result processing.
|
||||
log.error(
|
||||
f'Remote context error for {self.chan.uid}:{self.cid}:\n'
|
||||
f'{msg["error"]["tb_str"]}'
|
||||
)
|
||||
error = unpack_error(msg, self.chan)
|
||||
if (
|
||||
isinstance(error, ContextCancelled) and
|
||||
self._cancel_called
|
||||
):
|
||||
# this is an expected cancel request response message
|
||||
# and we don't need to raise it in scope since it will
|
||||
# potentially override a real error
|
||||
return
|
||||
|
||||
self._error = error
|
||||
|
||||
# TODO: tempted to **not** do this by-reraising in a
|
||||
# nursery and instead cancel a surrounding scope, detect
|
||||
# the cancellation, then lookup the error that was set?
|
||||
if self._scope_nursery:
|
||||
|
||||
async def raiser():
|
||||
raise self._error from None
|
||||
|
||||
# from trio.testing import wait_all_tasks_blocked
|
||||
# await wait_all_tasks_blocked()
|
||||
if not self._scope_nursery._closed: # type: ignore
|
||||
self._scope_nursery.start_soon(raiser)
|
||||
|
||||
async def cancel(
|
||||
self,
|
||||
msg: Optional[str] = None,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Cancel this inter-actor-task context.
|
||||
|
||||
Request that the far side cancel it's current linked context,
|
||||
Timeout quickly in an attempt to sidestep 2-generals...
|
||||
|
||||
'''
|
||||
side = 'caller' if self._portal else 'callee'
|
||||
if msg:
|
||||
assert side == 'callee', 'Only callee side can provide cancel msg'
|
||||
|
||||
log.cancel(f'Cancelling {side} side of context to {self.chan.uid}')
|
||||
|
||||
self._cancel_called = True
|
||||
|
||||
if side == 'caller':
|
||||
if not self._portal:
|
||||
raise RuntimeError(
|
||||
"No portal found, this is likely a callee side context"
|
||||
)
|
||||
|
||||
cid = self.cid
|
||||
with trio.move_on_after(0.5) as cs:
|
||||
cs.shield = True
|
||||
log.cancel(
|
||||
f"Cancelling stream {cid} to "
|
||||
f"{self._portal.channel.uid}")
|
||||
|
||||
# NOTE: we're telling the far end actor to cancel a task
|
||||
# corresponding to *this actor*. The far end local channel
|
||||
# instance is passed to `Actor._cancel_task()` implicitly.
|
||||
await self._portal.run_from_ns('self', '_cancel_task', cid=cid)
|
||||
|
||||
if cs.cancelled_caught:
|
||||
# XXX: there's no way to know if the remote task was indeed
|
||||
# cancelled in the case where the connection is broken or
|
||||
# some other network error occurred.
|
||||
# if not self._portal.channel.connected():
|
||||
if not self.chan.connected():
|
||||
log.cancel(
|
||||
"May have failed to cancel remote task "
|
||||
f"{cid} for {self._portal.channel.uid}")
|
||||
else:
|
||||
log.cancel(
|
||||
"Timed out on cancelling remote task "
|
||||
f"{cid} for {self._portal.channel.uid}")
|
||||
|
||||
# callee side remote task
|
||||
else:
|
||||
self._cancel_msg = msg
|
||||
|
||||
# TODO: should we have an explicit cancel message
|
||||
# or is relaying the local `trio.Cancelled` as an
|
||||
# {'error': trio.Cancelled, cid: "blah"} enough?
|
||||
# This probably gets into the discussion in
|
||||
# https://github.com/goodboy/tractor/issues/36
|
||||
assert self._scope_nursery
|
||||
self._scope_nursery.cancel_scope.cancel()
|
||||
|
||||
if self._recv_chan:
|
||||
await self._recv_chan.aclose()
|
||||
|
||||
@asynccontextmanager
|
||||
async def open_stream(
|
||||
|
||||
self,
|
||||
backpressure: Optional[bool] = True,
|
||||
msg_buffer_size: Optional[int] = None,
|
||||
|
||||
) -> AsyncGenerator[MsgStream, None]:
|
||||
'''
|
||||
Open a ``MsgStream``, a bi-directional stream connected to the
|
||||
cross-actor (far end) task for this ``Context``.
|
||||
|
||||
This context manager must be entered on both the caller and
|
||||
callee for the stream to logically be considered "connected".
|
||||
|
||||
A ``MsgStream`` is currently "one-shot" use, meaning if you
|
||||
close it you can not "re-open" it for streaming and instead you
|
||||
must re-establish a new surrounding ``Context`` using
|
||||
``Portal.open_context()``. In the future this may change but
|
||||
currently there seems to be no obvious reason to support
|
||||
"re-opening":
|
||||
- pausing a stream can be done with a message.
|
||||
- task errors will normally require a restart of the entire
|
||||
scope of the inter-actor task context due to the nature of
|
||||
``trio``'s cancellation system.
|
||||
|
||||
'''
|
||||
actor = current_actor()
|
||||
|
||||
# here we create a mem chan that corresponds to the
|
||||
# far end caller / callee.
|
||||
|
||||
# Likewise if the surrounding context has been cancelled we error here
|
||||
# since it likely means the surrounding block was exited or
|
||||
# killed
|
||||
|
||||
if self._cancel_called:
|
||||
task = trio.lowlevel.current_task().name
|
||||
raise ContextCancelled(
|
||||
f'Context around {actor.uid[0]}:{task} was already cancelled!'
|
||||
)
|
||||
|
||||
if not self._portal and not self._started_called:
|
||||
raise RuntimeError(
|
||||
'Context.started()` must be called before opening a stream'
|
||||
)
|
||||
|
||||
# NOTE: in one way streaming this only happens on the
|
||||
# caller side inside `Actor.start_remote_task()` so if you try
|
||||
# to send a stop from the caller to the callee in the
|
||||
# single-direction-stream case you'll get a lookup error
|
||||
# currently.
|
||||
ctx = actor.get_context(
|
||||
self.chan,
|
||||
self.cid,
|
||||
msg_buffer_size=msg_buffer_size,
|
||||
)
|
||||
ctx._backpressure = backpressure
|
||||
assert ctx is self
|
||||
|
||||
# XXX: If the underlying channel feeder receive mem chan has
|
||||
# been closed then likely client code has already exited
|
||||
# a ``.open_stream()`` block prior or there was some other
|
||||
# unanticipated error or cancellation from ``trio``.
|
||||
|
||||
if ctx._recv_chan._closed:
|
||||
raise trio.ClosedResourceError(
|
||||
'The underlying channel for this stream was already closed!?')
|
||||
|
||||
async with MsgStream(
|
||||
ctx=self,
|
||||
rx_chan=ctx._recv_chan,
|
||||
) as stream:
|
||||
|
||||
if self._portal:
|
||||
self._portal._streams.add(stream)
|
||||
|
||||
try:
|
||||
self._stream_opened = True
|
||||
|
||||
# XXX: do we need this?
|
||||
# ensure we aren't cancelled before yielding the stream
|
||||
# await trio.lowlevel.checkpoint()
|
||||
yield stream
|
||||
|
||||
# NOTE: Make the stream "one-shot use". On exit, signal
|
||||
# ``trio.EndOfChannel``/``StopAsyncIteration`` to the
|
||||
# far end.
|
||||
await stream.aclose()
|
||||
|
||||
finally:
|
||||
if self._portal:
|
||||
try:
|
||||
self._portal._streams.remove(stream)
|
||||
except KeyError:
|
||||
log.warning(
|
||||
f'Stream was already destroyed?\n'
|
||||
f'actor: {self.chan.uid}\n'
|
||||
f'ctx id: {self.cid}'
|
||||
)
|
||||
|
||||
async def result(self) -> Any:
|
||||
'''
|
||||
From a caller side, wait for and return the final result from
|
||||
the callee side task.
|
||||
|
||||
'''
|
||||
assert self._portal, "Context.result() can not be called from callee!"
|
||||
assert self._recv_chan
|
||||
|
||||
if self._result is False:
|
||||
|
||||
if not self._recv_chan._closed: # type: ignore
|
||||
|
||||
# wait for a final context result consuming
|
||||
# and discarding any bi dir stream msgs still
|
||||
# in transit from the far end.
|
||||
while True:
|
||||
|
||||
msg = await self._recv_chan.receive()
|
||||
try:
|
||||
self._result = msg['return']
|
||||
break
|
||||
except KeyError as msgerr:
|
||||
|
||||
if 'yield' in msg:
|
||||
# far end task is still streaming to us so discard
|
||||
log.warning(f'Discarding stream delivered {msg}')
|
||||
continue
|
||||
|
||||
elif 'stop' in msg:
|
||||
log.debug('Remote stream terminated')
|
||||
continue
|
||||
|
||||
# internal error should never get here
|
||||
assert msg.get('cid'), (
|
||||
"Received internal error at portal?")
|
||||
|
||||
raise unpack_error(
|
||||
msg, self._portal.channel
|
||||
) from msgerr
|
||||
|
||||
return self._result
|
||||
|
||||
async def started(
|
||||
self,
|
||||
value: Optional[Any] = None
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Indicate to calling actor's task that this linked context
|
||||
has started and send ``value`` to the other side.
|
||||
|
||||
On the calling side ``value`` is the second item delivered
|
||||
in the tuple returned by ``Portal.open_context()``.
|
||||
|
||||
'''
|
||||
if self._portal:
|
||||
raise RuntimeError(
|
||||
f"Caller side context {self} can not call started!")
|
||||
|
||||
elif self._started_called:
|
||||
raise RuntimeError(
|
||||
f"called 'started' twice on context with {self.chan.uid}")
|
||||
|
||||
await self.chan.send({'started': value, 'cid': self.cid})
|
||||
self._started_called = True
|
||||
|
||||
# TODO: do we need a restart api?
|
||||
# async def restart(self) -> None:
|
||||
# pass
|
||||
|
||||
|
||||
def stream(func: Callable) -> Callable:
|
||||
"""Mark an async function as a streaming routine with ``@stream``.
|
||||
|
||||
"""
|
||||
# annotate
|
||||
# TODO: apply whatever solution ``mypy`` ends up picking for this:
|
||||
# https://github.com/python/mypy/issues/2087#issuecomment-769266912
|
||||
func._tractor_stream_function = True # type: ignore
|
||||
|
|
@ -379,3 +734,22 @@ def stream(func: Callable) -> Callable:
|
|||
"(Or ``to_trio`` if using ``asyncio`` in guest mode)."
|
||||
)
|
||||
return func
|
||||
|
||||
|
||||
def context(func: Callable) -> Callable:
|
||||
"""Mark an async function as a streaming routine with ``@context``.
|
||||
|
||||
"""
|
||||
# annotate
|
||||
# TODO: apply whatever solution ``mypy`` ends up picking for this:
|
||||
# https://github.com/python/mypy/issues/2087#issuecomment-769266912
|
||||
func._tractor_context_function = True # type: ignore
|
||||
|
||||
sig = inspect.signature(func)
|
||||
params = sig.parameters
|
||||
if 'ctx' not in params:
|
||||
raise TypeError(
|
||||
"The first argument to the context function "
|
||||
f"{func.__name__} must be `ctx: tractor.Context`"
|
||||
)
|
||||
return func
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ import trio
|
|||
import wrapt
|
||||
|
||||
from ..log import get_logger
|
||||
from .._context import Context
|
||||
from .._streaming import Context
|
||||
|
||||
|
||||
__all__ = ['pub']
|
||||
|
|
@ -148,8 +148,7 @@ def pub(
|
|||
*,
|
||||
tasks: set[str] = set(),
|
||||
):
|
||||
'''
|
||||
Publisher async generator decorator.
|
||||
"""Publisher async generator decorator.
|
||||
|
||||
A publisher can be called multiple times from different actors but
|
||||
will only spawn a finite set of internal tasks to stream values to
|
||||
|
|
@ -228,8 +227,7 @@ def pub(
|
|||
running in a single actor to stream data to an arbitrary number of
|
||||
subscribers. If you are ok to have a new task running for every call
|
||||
to ``pub_service()`` then probably don't need this.
|
||||
|
||||
'''
|
||||
"""
|
||||
global _pubtask2lock
|
||||
|
||||
# handle the decorator not called with () case
|
||||
|
|
|
|||
|
|
@ -82,10 +82,6 @@ class StackLevelAdapter(logging.LoggerAdapter):
|
|||
msg: str,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
IPC level msg-ing.
|
||||
|
||||
'''
|
||||
return self.log(5, msg)
|
||||
|
||||
def runtime(
|
||||
|
|
@ -98,20 +94,12 @@ class StackLevelAdapter(logging.LoggerAdapter):
|
|||
self,
|
||||
msg: str,
|
||||
) -> None:
|
||||
'''
|
||||
Cancellation logging, mostly for runtime reporting.
|
||||
|
||||
'''
|
||||
return self.log(16, msg)
|
||||
|
||||
def pdb(
|
||||
self,
|
||||
msg: str,
|
||||
) -> None:
|
||||
'''
|
||||
Debugger logging.
|
||||
|
||||
'''
|
||||
return self.log(500, msg)
|
||||
|
||||
def log(self, level, msg, *args, **kwargs):
|
||||
|
|
|
|||
|
|
@ -25,16 +25,8 @@ from collections import deque
|
|||
from contextlib import asynccontextmanager
|
||||
from functools import partial
|
||||
from operator import ne
|
||||
from typing import (
|
||||
Optional,
|
||||
Callable,
|
||||
Awaitable,
|
||||
Any,
|
||||
AsyncIterator,
|
||||
Protocol,
|
||||
Generic,
|
||||
TypeVar,
|
||||
)
|
||||
from typing import Optional, Callable, Awaitable, Any, AsyncIterator, Protocol
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
import trio
|
||||
from trio._core._run import Task
|
||||
|
|
|
|||
|
|
@ -237,7 +237,7 @@ async def maybe_open_context(
|
|||
yielded = _Cache.values[ctx_key]
|
||||
|
||||
except KeyError:
|
||||
log.debug(f'Allocating new {acm_func} for {ctx_key}')
|
||||
log.info(f'Allocating new {acm_func} for {ctx_key}')
|
||||
mngr = acm_func(**kwargs)
|
||||
resources = _Cache.resources
|
||||
assert not resources.get(ctx_key), f'Resource exists? {ctx_key}'
|
||||
|
|
@ -265,7 +265,7 @@ async def maybe_open_context(
|
|||
if yielded is not None:
|
||||
# if no more consumers, teardown the client
|
||||
if _Cache.users <= 0:
|
||||
log.debug(f'De-allocating resource for {ctx_key}')
|
||||
log.info(f'De-allocating resource for {ctx_key}')
|
||||
|
||||
# XXX: if we're cancelled we the entry may have never
|
||||
# been entered since the nursery task was killed.
|
||||
|
|
|
|||
Loading…
Reference in New Issue