Compare commits
	
		
			No commits in common. "main" and "cancelled_masking_guards" have entirely different histories. 
		
	
	
		
			main
			...
			cancelled_
		
	
		|  | @ -24,10 +24,14 @@ from tractor._testing import ( | ||||||
| ) | ) | ||||||
| 
 | 
 | ||||||
| # XXX TODO cases: | # XXX TODO cases: | ||||||
|  | # - [ ] peer cancelled itself - so other peers should | ||||||
|  | #   get errors reflecting that the peer was itself the .canceller? | ||||||
|  | 
 | ||||||
| # - [x] WE cancelled the peer and thus should not see any raised | # - [x] WE cancelled the peer and thus should not see any raised | ||||||
| #   `ContextCancelled` as it should be reaped silently? | #   `ContextCancelled` as it should be reaped silently? | ||||||
| #   => pretty sure `test_context_stream_semantics::test_caller_cancels()` | #   => pretty sure `test_context_stream_semantics::test_caller_cancels()` | ||||||
| #      already covers this case? | #      already covers this case? | ||||||
|  | 
 | ||||||
| # - [x] INTER-PEER: some arbitrary remote peer cancels via | # - [x] INTER-PEER: some arbitrary remote peer cancels via | ||||||
| #   Portal.cancel_actor(). | #   Portal.cancel_actor(). | ||||||
| #   => all other connected peers should get that cancel requesting peer's | #   => all other connected peers should get that cancel requesting peer's | ||||||
|  | @ -40,6 +44,16 @@ from tractor._testing import ( | ||||||
| #   that also spawned a remote task task in that same peer-parent. | #   that also spawned a remote task task in that same peer-parent. | ||||||
| 
 | 
 | ||||||
| 
 | 
 | ||||||
|  | # def test_self_cancel(): | ||||||
|  | #     ''' | ||||||
|  | #     2 cases: | ||||||
|  | #     - calls `Actor.cancel()` locally in some task | ||||||
|  | #     - calls LocalPortal.cancel_actor()` ? | ||||||
|  | 
 | ||||||
|  | #     ''' | ||||||
|  | #     ... | ||||||
|  | 
 | ||||||
|  | 
 | ||||||
| @tractor.context | @tractor.context | ||||||
| async def open_stream_then_sleep_forever( | async def open_stream_then_sleep_forever( | ||||||
|     ctx: Context, |     ctx: Context, | ||||||
|  | @ -792,7 +806,7 @@ async def basic_echo_server( | ||||||
|     ctx: Context, |     ctx: Context, | ||||||
|     peer_name: str = 'wittle_bruv', |     peer_name: str = 'wittle_bruv', | ||||||
| 
 | 
 | ||||||
|     err_after_imsg: int|None = None, |     err_after: int|None = None, | ||||||
| 
 | 
 | ||||||
| ) -> None: | ) -> None: | ||||||
|     ''' |     ''' | ||||||
|  | @ -821,9 +835,8 @@ async def basic_echo_server( | ||||||
|             await ipc.send(resp) |             await ipc.send(resp) | ||||||
| 
 | 
 | ||||||
|             if ( |             if ( | ||||||
|                 err_after_imsg |                 err_after | ||||||
|                 and |                 and i > err_after | ||||||
|                 i > err_after_imsg |  | ||||||
|             ): |             ): | ||||||
|                 raise RuntimeError( |                 raise RuntimeError( | ||||||
|                     f'Simulated error in `{peer_name}`' |                     f'Simulated error in `{peer_name}`' | ||||||
|  | @ -965,8 +978,7 @@ async def tell_little_bro( | ||||||
|     actor_name: str, |     actor_name: str, | ||||||
| 
 | 
 | ||||||
|     caller: str = '', |     caller: str = '', | ||||||
|     err_after: float|None = None, |     err_after: int|None = None, | ||||||
|     rng_seed: int = 50, |  | ||||||
| ): | ): | ||||||
|     # contact target actor, do a stream dialog. |     # contact target actor, do a stream dialog. | ||||||
|     async with ( |     async with ( | ||||||
|  | @ -977,18 +989,14 @@ async def tell_little_bro( | ||||||
|             basic_echo_server, |             basic_echo_server, | ||||||
| 
 | 
 | ||||||
|             # XXX proxy any delayed err condition |             # XXX proxy any delayed err condition | ||||||
|             err_after_imsg=( |             err_after=err_after, | ||||||
|                 err_after * rng_seed |  | ||||||
|                 if err_after is not None |  | ||||||
|                 else None |  | ||||||
|             ), |  | ||||||
|         ) as (sub_ctx, first), |         ) as (sub_ctx, first), | ||||||
| 
 | 
 | ||||||
|         sub_ctx.open_stream() as echo_ipc, |         sub_ctx.open_stream() as echo_ipc, | ||||||
|     ): |     ): | ||||||
|         actor: Actor = current_actor() |         actor: Actor = current_actor() | ||||||
|         uid: tuple = actor.uid |         uid: tuple = actor.uid | ||||||
|         for i in range(rng_seed): |         for i in range(100): | ||||||
|             msg: tuple = ( |             msg: tuple = ( | ||||||
|                 uid, |                 uid, | ||||||
|                 i, |                 i, | ||||||
|  | @ -1013,13 +1021,13 @@ async def tell_little_bro( | ||||||
| ) | ) | ||||||
| @pytest.mark.parametrize( | @pytest.mark.parametrize( | ||||||
|     'raise_sub_spawn_error_after', |     'raise_sub_spawn_error_after', | ||||||
|     [None, 0.5], |     [None, 50], | ||||||
| ) | ) | ||||||
| def test_peer_spawns_and_cancels_service_subactor( | def test_peer_spawns_and_cancels_service_subactor( | ||||||
|     debug_mode: bool, |     debug_mode: bool, | ||||||
|     raise_client_error: str, |     raise_client_error: str, | ||||||
|     reg_addr: tuple[str, int], |     reg_addr: tuple[str, int], | ||||||
|     raise_sub_spawn_error_after: float|None, |     raise_sub_spawn_error_after: int|None, | ||||||
| ): | ): | ||||||
|     # NOTE: this tests for the modden `mod wks open piker` bug |     # NOTE: this tests for the modden `mod wks open piker` bug | ||||||
|     # discovered as part of implementing workspace ctx |     # discovered as part of implementing workspace ctx | ||||||
|  | @ -1033,7 +1041,6 @@ def test_peer_spawns_and_cancels_service_subactor( | ||||||
|     #   and the server's spawned child should cancel and terminate! |     #   and the server's spawned child should cancel and terminate! | ||||||
|     peer_name: str = 'little_bro' |     peer_name: str = 'little_bro' | ||||||
| 
 | 
 | ||||||
| 
 |  | ||||||
|     def check_inner_rte(rae: RemoteActorError): |     def check_inner_rte(rae: RemoteActorError): | ||||||
|         ''' |         ''' | ||||||
|         Validate the little_bro's relayed inception! |         Validate the little_bro's relayed inception! | ||||||
|  | @ -1127,7 +1134,8 @@ def test_peer_spawns_and_cancels_service_subactor( | ||||||
|                         ) |                         ) | ||||||
| 
 | 
 | ||||||
|                     try: |                     try: | ||||||
|                         res = await client_ctx.wait_for_result(hide_tb=False) |                         res = await client_ctx.result(hide_tb=False) | ||||||
|  | 
 | ||||||
|                         # in remote (relayed inception) error |                         # in remote (relayed inception) error | ||||||
|                         # case, we should error on the line above! |                         # case, we should error on the line above! | ||||||
|                         if raise_sub_spawn_error_after: |                         if raise_sub_spawn_error_after: | ||||||
|  | @ -1138,23 +1146,6 @@ def test_peer_spawns_and_cancels_service_subactor( | ||||||
|                         assert isinstance(res, ContextCancelled) |                         assert isinstance(res, ContextCancelled) | ||||||
|                         assert client_ctx.cancel_acked |                         assert client_ctx.cancel_acked | ||||||
|                         assert res.canceller == root.uid |                         assert res.canceller == root.uid | ||||||
|                         assert not raise_sub_spawn_error_after |  | ||||||
| 
 |  | ||||||
|                         # cancelling the spawner sub should |  | ||||||
|                         # transitively cancel it's sub, the little |  | ||||||
|                         # bruv. |  | ||||||
|                         print('root cancelling server/client sub-actors') |  | ||||||
|                         await spawn_ctx.cancel() |  | ||||||
|                         async with tractor.find_actor( |  | ||||||
|                             name=peer_name, |  | ||||||
|                         ) as sub: |  | ||||||
|                             assert not sub |  | ||||||
| 
 |  | ||||||
|                     # XXX, only for tracing |  | ||||||
|                     # except BaseException as _berr: |  | ||||||
|                     #     berr = _berr |  | ||||||
|                     #     await tractor.pause(shield=True) |  | ||||||
|                     #     raise berr |  | ||||||
| 
 | 
 | ||||||
|                     except RemoteActorError as rae: |                     except RemoteActorError as rae: | ||||||
|                         _err = rae |                         _err = rae | ||||||
|  | @ -1183,8 +1174,19 @@ def test_peer_spawns_and_cancels_service_subactor( | ||||||
|                         raise |                         raise | ||||||
|                         # await tractor.pause() |                         # await tractor.pause() | ||||||
| 
 | 
 | ||||||
|  |                     else: | ||||||
|  |                         assert not raise_sub_spawn_error_after | ||||||
|  | 
 | ||||||
|  |                         # cancelling the spawner sub should | ||||||
|  |                         # transitively cancel it's sub, the little | ||||||
|  |                         # bruv. | ||||||
|  |                         print('root cancelling server/client sub-actors') | ||||||
|  |                         await spawn_ctx.cancel() | ||||||
|  |                         async with tractor.find_actor( | ||||||
|  |                             name=peer_name, | ||||||
|  |                         ) as sub: | ||||||
|  |                             assert not sub | ||||||
| 
 | 
 | ||||||
|                     # await tractor.pause() |  | ||||||
|                     # await server.cancel_actor() |                     # await server.cancel_actor() | ||||||
| 
 | 
 | ||||||
|             except RemoteActorError as rae: |             except RemoteActorError as rae: | ||||||
|  | @ -1197,7 +1199,7 @@ def test_peer_spawns_and_cancels_service_subactor( | ||||||
| 
 | 
 | ||||||
|             # since we called `.cancel_actor()`, `.cancel_ack` |             # since we called `.cancel_actor()`, `.cancel_ack` | ||||||
|             # will not be set on the ctx bc `ctx.cancel()` was not |             # will not be set on the ctx bc `ctx.cancel()` was not | ||||||
|             # called directly for this confext. |             # called directly fot this confext. | ||||||
|             except ContextCancelled as ctxc: |             except ContextCancelled as ctxc: | ||||||
|                 _ctxc = ctxc |                 _ctxc = ctxc | ||||||
|                 print( |                 print( | ||||||
|  | @ -1237,19 +1239,12 @@ def test_peer_spawns_and_cancels_service_subactor( | ||||||
| 
 | 
 | ||||||
|                 # assert spawn_ctx.cancelled_caught |                 # assert spawn_ctx.cancelled_caught | ||||||
| 
 | 
 | ||||||
|     async def _main(): |  | ||||||
|         with trio.fail_after( |  | ||||||
|             3 if not debug_mode |  | ||||||
|             else 999 |  | ||||||
|         ): |  | ||||||
|             await main() |  | ||||||
| 
 |  | ||||||
|     if raise_sub_spawn_error_after: |     if raise_sub_spawn_error_after: | ||||||
|         with pytest.raises(RemoteActorError) as excinfo: |         with pytest.raises(RemoteActorError) as excinfo: | ||||||
|             trio.run(_main) |             trio.run(main) | ||||||
| 
 | 
 | ||||||
|         rae: RemoteActorError = excinfo.value |         rae: RemoteActorError = excinfo.value | ||||||
|         check_inner_rte(rae) |         check_inner_rte(rae) | ||||||
| 
 | 
 | ||||||
|     else: |     else: | ||||||
|         trio.run(_main) |         trio.run(main) | ||||||
|  |  | ||||||
|  | @ -1,239 +0,0 @@ | ||||||
| ''' |  | ||||||
| Define the details of inter-actor "out-of-band" (OoB) cancel |  | ||||||
| semantics, that is how cancellation works when a cancel request comes |  | ||||||
| from the different concurrency (primitive's) "layer" then where the |  | ||||||
| eventual `trio.Task` actually raises a signal. |  | ||||||
| 
 |  | ||||||
| ''' |  | ||||||
| from functools import partial |  | ||||||
| # from contextlib import asynccontextmanager as acm |  | ||||||
| # import itertools |  | ||||||
| 
 |  | ||||||
| import pytest |  | ||||||
| import trio |  | ||||||
| import tractor |  | ||||||
| from tractor import (  # typing |  | ||||||
|     ActorNursery, |  | ||||||
|     Portal, |  | ||||||
|     Context, |  | ||||||
|     # ContextCancelled, |  | ||||||
|     # RemoteActorError, |  | ||||||
| ) |  | ||||||
| # from tractor._testing import ( |  | ||||||
| #     tractor_test, |  | ||||||
| #     expect_ctxc, |  | ||||||
| # ) |  | ||||||
| 
 |  | ||||||
| # XXX TODO cases: |  | ||||||
| # - [ ] peer cancelled itself - so other peers should |  | ||||||
| #   get errors reflecting that the peer was itself the .canceller? |  | ||||||
| 
 |  | ||||||
| # def test_self_cancel(): |  | ||||||
| #     ''' |  | ||||||
| #     2 cases: |  | ||||||
| #     - calls `Actor.cancel()` locally in some task |  | ||||||
| #     - calls LocalPortal.cancel_actor()` ? |  | ||||||
| # |  | ||||||
| # things to ensure! |  | ||||||
| # -[ ] the ctxc raised in a child should ideally show the tb of the |  | ||||||
| #     underlying `Cancelled` checkpoint, i.e. |  | ||||||
| #     `raise scope_error from ctxc`? |  | ||||||
| # |  | ||||||
| # -[ ] a self-cancelled context, if not allowed to block on |  | ||||||
| #     `ctx.result()` at some point will hang since the `ctx._scope` |  | ||||||
| #     is never `.cancel_called`; cases for this include, |  | ||||||
| #     - an `open_ctx()` which never starteds before being OoB actor |  | ||||||
| #       cancelled. |  | ||||||
| #       |_ parent task will be blocked in `.open_context()` for the |  | ||||||
| #         `Started` msg, and when the OoB ctxc arrives `ctx._scope` |  | ||||||
| #         will never have been signalled.. |  | ||||||
| 
 |  | ||||||
| #     ''' |  | ||||||
| #     ... |  | ||||||
| 
 |  | ||||||
| # TODO, sanity test against the case in `/examples/trio/lockacquire_not_unmasked.py` |  | ||||||
| # but with the `Lock.acquire()` from a `@context` to ensure the |  | ||||||
| # implicit ignore-case-non-unmasking. |  | ||||||
| # |  | ||||||
| # @tractor.context |  | ||||||
| # async def acquire_actor_global_lock( |  | ||||||
| #     ctx: tractor.Context, |  | ||||||
| #     ignore_special_cases: bool, |  | ||||||
| # ): |  | ||||||
| 
 |  | ||||||
| #     async with maybe_unmask_excs( |  | ||||||
| #         ignore_special_cases=ignore_special_cases, |  | ||||||
| #     ): |  | ||||||
| #         await ctx.started('locked') |  | ||||||
| 
 |  | ||||||
| #     # block til cancelled |  | ||||||
| #     await trio.sleep_forever() |  | ||||||
| 
 |  | ||||||
| 
 |  | ||||||
| @tractor.context |  | ||||||
| async def sleep_forever( |  | ||||||
|     ctx: tractor.Context, |  | ||||||
|     # ignore_special_cases: bool, |  | ||||||
|     do_started: bool, |  | ||||||
| ): |  | ||||||
| 
 |  | ||||||
|     # async with maybe_unmask_excs( |  | ||||||
|     #     ignore_special_cases=ignore_special_cases, |  | ||||||
|     # ): |  | ||||||
|     #     await ctx.started('locked') |  | ||||||
|     if do_started: |  | ||||||
|         await ctx.started() |  | ||||||
| 
 |  | ||||||
|     # block til cancelled |  | ||||||
|     print('sleepin on child-side..') |  | ||||||
|     await trio.sleep_forever() |  | ||||||
| 
 |  | ||||||
| 
 |  | ||||||
| @pytest.mark.parametrize( |  | ||||||
|     'cancel_ctx', |  | ||||||
|     [True, False], |  | ||||||
| ) |  | ||||||
| def test_cancel_ctx_with_parent_side_entered_in_bg_task( |  | ||||||
|     debug_mode: bool, |  | ||||||
|     loglevel: str, |  | ||||||
|     cancel_ctx: bool, |  | ||||||
| ): |  | ||||||
|     ''' |  | ||||||
|     The most "basic" out-of-band-task self-cancellation case where |  | ||||||
|     `Portal.open_context()` is entered in a bg task and the |  | ||||||
|     parent-task (of the containing nursery) calls `Context.cancel()` |  | ||||||
|     without the child knowing; the `Context._scope` should be |  | ||||||
|     `.cancel_called` when the IPC ctx's child-side relays |  | ||||||
|     a `ContextCancelled` with a `.canceller` set to the parent |  | ||||||
|     actor('s task). |  | ||||||
| 
 |  | ||||||
|     ''' |  | ||||||
|     async def main(): |  | ||||||
|         with trio.fail_after( |  | ||||||
|             2 if not debug_mode else 999, |  | ||||||
|         ): |  | ||||||
|             an: ActorNursery |  | ||||||
|             async with ( |  | ||||||
|                 tractor.open_nursery( |  | ||||||
|                     debug_mode=debug_mode, |  | ||||||
|                     loglevel='devx', |  | ||||||
|                     enable_stack_on_sig=True, |  | ||||||
|                 ) as an, |  | ||||||
|                 trio.open_nursery() as tn, |  | ||||||
|             ): |  | ||||||
|                 ptl: Portal = await an.start_actor( |  | ||||||
|                     'sub', |  | ||||||
|                     enable_modules=[__name__], |  | ||||||
|                 ) |  | ||||||
| 
 |  | ||||||
|                 async def _open_ctx_async( |  | ||||||
|                     do_started: bool = True, |  | ||||||
|                     task_status=trio.TASK_STATUS_IGNORED, |  | ||||||
|                 ): |  | ||||||
|                     # do we expect to never enter the |  | ||||||
|                     # `.open_context()` below. |  | ||||||
|                     if not do_started: |  | ||||||
|                         task_status.started() |  | ||||||
| 
 |  | ||||||
|                     async with ptl.open_context( |  | ||||||
|                         sleep_forever, |  | ||||||
|                         do_started=do_started, |  | ||||||
|                     ) as (ctx, first): |  | ||||||
|                         task_status.started(ctx) |  | ||||||
|                         await trio.sleep_forever() |  | ||||||
| 
 |  | ||||||
|                 # XXX, this is the key OoB part! |  | ||||||
|                 # |  | ||||||
|                 # - start the `.open_context()` in a bg task which |  | ||||||
|                 #   blocks inside the embedded scope-body, |  | ||||||
|                 # |  | ||||||
|                 # -  when we call `Context.cancel()` it **is |  | ||||||
|                 #   not** from the same task which eventually runs |  | ||||||
|                 #   `.__aexit__()`, |  | ||||||
|                 # |  | ||||||
|                 # - since the bg "opener" task will be in |  | ||||||
|                 #   a `trio.sleep_forever()`, it must be interrupted |  | ||||||
|                 #   by the `ContextCancelled` delivered from the |  | ||||||
|                 #   child-side; `Context._scope: CancelScope` MUST |  | ||||||
|                 #   be `.cancel_called`! |  | ||||||
|                 # |  | ||||||
|                 print('ASYNC opening IPC context in subtask..') |  | ||||||
|                 maybe_ctx: Context|None = await tn.start(partial( |  | ||||||
|                     _open_ctx_async, |  | ||||||
|                 )) |  | ||||||
| 
 |  | ||||||
|                 if ( |  | ||||||
|                     maybe_ctx |  | ||||||
|                     and |  | ||||||
|                     cancel_ctx |  | ||||||
|                 ): |  | ||||||
|                     print('cancelling first IPC ctx!') |  | ||||||
|                     await maybe_ctx.cancel() |  | ||||||
| 
 |  | ||||||
|                 # XXX, note that despite `maybe_context.cancel()` |  | ||||||
|                 # being called above, it's the parent (bg) task |  | ||||||
|                 # which was originally never interrupted in |  | ||||||
|                 # the `ctx._scope` body due to missing case logic in |  | ||||||
|                 # `ctx._maybe_cancel_and_set_remote_error()`. |  | ||||||
|                 # |  | ||||||
|                 # It didn't matter that the subactor process was |  | ||||||
|                 # already terminated and reaped, nothing was |  | ||||||
|                 # cancelling the ctx-parent task's scope! |  | ||||||
|                 # |  | ||||||
|                 print('cancelling subactor!') |  | ||||||
|                 await ptl.cancel_actor() |  | ||||||
| 
 |  | ||||||
|                 if maybe_ctx: |  | ||||||
|                     try: |  | ||||||
|                         await maybe_ctx.wait_for_result() |  | ||||||
|                     except tractor.ContextCancelled as ctxc: |  | ||||||
|                         assert not cancel_ctx |  | ||||||
|                         assert ( |  | ||||||
|                             ctxc.canceller |  | ||||||
|                             == |  | ||||||
|                             tractor.current_actor().aid.uid |  | ||||||
|                         ) |  | ||||||
|                         # don't re-raise since it'll trigger |  | ||||||
|                         # an EG from the above tn. |  | ||||||
| 
 |  | ||||||
|     if cancel_ctx: |  | ||||||
|         # graceful self-cancel |  | ||||||
|         trio.run(main) |  | ||||||
| 
 |  | ||||||
|     else: |  | ||||||
|         # ctx parent task should see OoB ctxc due to |  | ||||||
|         # `ptl.cancel_actor()`. |  | ||||||
|         with pytest.raises(tractor.ContextCancelled) as excinfo: |  | ||||||
|             trio.run(main) |  | ||||||
| 
 |  | ||||||
|         assert 'root' in excinfo.value.canceller[0] |  | ||||||
| 
 |  | ||||||
| 
 |  | ||||||
| # def test_parent_actor_cancels_subactor_with_gt1_ctxs_open_to_it( |  | ||||||
| #     debug_mode: bool, |  | ||||||
| #     loglevel: str, |  | ||||||
| # ): |  | ||||||
| #     ''' |  | ||||||
| #     Demos OoB cancellation from the perspective of a ctx opened with |  | ||||||
| #     a child subactor where the parent cancels the child at the "actor |  | ||||||
| #     layer" using `Portal.cancel_actor()` and thus the |  | ||||||
| #     `ContextCancelled.canceller` received by the ctx's parent-side |  | ||||||
| #     task will appear to be a "self cancellation" even though that |  | ||||||
| #     specific task itself was not cancelled and thus |  | ||||||
| #     `Context.cancel_called ==False`. |  | ||||||
| #     ''' |  | ||||||
|                 # TODO, do we have an existing implied ctx |  | ||||||
|                 # cancel test like this? |  | ||||||
|                 # with trio.move_on_after(0.5):# as cs: |  | ||||||
|                 #     await _open_ctx_async( |  | ||||||
|                 #         do_started=False, |  | ||||||
|                 #     ) |  | ||||||
| 
 |  | ||||||
| 
 |  | ||||||
|                 # in-line ctx scope should definitely raise |  | ||||||
|                 # a ctxc with `.canceller = 'root'` |  | ||||||
|                 # async with ptl.open_context( |  | ||||||
|                 #     sleep_forever, |  | ||||||
|                 #     do_started=True, |  | ||||||
|                 # ) as pair: |  | ||||||
| 
 |  | ||||||
|  | @ -279,7 +279,7 @@ def test_unmask_aclose_as_checkpoint_on_aexit( | ||||||
|         ), |         ), | ||||||
|     ] |     ] | ||||||
| ) | ) | ||||||
| def test_cancelled_lockacquire_in_ipctx_not_unmasked( | def test_cancelled_lockacquire_in_ipctx_not_unmaskeed( | ||||||
|     ignore_special_cases: bool, |     ignore_special_cases: bool, | ||||||
|     loglevel: str, |     loglevel: str, | ||||||
|     debug_mode: bool, |     debug_mode: bool, | ||||||
|  |  | ||||||
|  | @ -442,25 +442,25 @@ class Context: | ||||||
|         ''' |         ''' | ||||||
|         Records whether cancellation has been requested for this context |         Records whether cancellation has been requested for this context | ||||||
|         by a call to  `.cancel()` either due to, |         by a call to  `.cancel()` either due to, | ||||||
|         - an explicit call by some local task, |         - either an explicit call by some local task, | ||||||
|         - or an implicit call due to an error caught inside |         - or an implicit call due to an error caught inside | ||||||
|           the `Portal.open_context()` block. |           the ``Portal.open_context()`` block. | ||||||
| 
 | 
 | ||||||
|         ''' |         ''' | ||||||
|         return self._cancel_called |         return self._cancel_called | ||||||
| 
 | 
 | ||||||
|     # XXX, to debug who frickin sets it.. |     @cancel_called.setter | ||||||
|     # @cancel_called.setter |     def cancel_called(self, val: bool) -> None: | ||||||
|     # def cancel_called(self, val: bool) -> None: |         ''' | ||||||
|     #     ''' |         Set the self-cancelled request `bool` value. | ||||||
|     #     Set the self-cancelled request `bool` value. |  | ||||||
| 
 | 
 | ||||||
|     #     ''' |         ''' | ||||||
|  |         # to debug who frickin sets it.. | ||||||
|         # if val: |         # if val: | ||||||
|         #     from .devx import pause_from_sync |         #     from .devx import pause_from_sync | ||||||
|         #     pause_from_sync() |         #     pause_from_sync() | ||||||
| 
 | 
 | ||||||
|     #     self._cancel_called = val |         self._cancel_called = val | ||||||
| 
 | 
 | ||||||
|     @property |     @property | ||||||
|     def canceller(self) -> tuple[str, str]|None: |     def canceller(self) -> tuple[str, str]|None: | ||||||
|  | @ -635,71 +635,6 @@ class Context: | ||||||
|         ''' |         ''' | ||||||
|         await self.chan.send(Stop(cid=self.cid)) |         await self.chan.send(Stop(cid=self.cid)) | ||||||
| 
 | 
 | ||||||
|     @property |  | ||||||
|     def parent_task(self) -> trio.Task: |  | ||||||
|         ''' |  | ||||||
|         This IPC context's "owning task" which is a `trio.Task` |  | ||||||
|         on one of the "sides" of the IPC. |  | ||||||
| 
 |  | ||||||
|         Note that the "parent_" prefix here refers to the local |  | ||||||
|         `trio` task tree using the same interface as |  | ||||||
|         `trio.Nursery.parent_task` whereas for IPC contexts, |  | ||||||
|         a different cross-actor task hierarchy exists: |  | ||||||
| 
 |  | ||||||
|         - a "parent"-side which originally entered |  | ||||||
|           `Portal.open_context()`, |  | ||||||
| 
 |  | ||||||
|         - the "child"-side which was spawned and scheduled to invoke |  | ||||||
|           a function decorated with `@tractor.context`. |  | ||||||
| 
 |  | ||||||
|         This task is thus a handle to mem-domain-distinct/per-process |  | ||||||
|         `Nursery.parent_task` depending on in which of the above |  | ||||||
|         "sides" this context exists. |  | ||||||
| 
 |  | ||||||
|         ''' |  | ||||||
|         return self._task |  | ||||||
| 
 |  | ||||||
|     def _is_blocked_on_rx_chan(self) -> bool: |  | ||||||
|         ''' |  | ||||||
|         Predicate to indicate whether the owner `._task: trio.Task` is |  | ||||||
|         currently blocked (by `.receive()`-ing) on its underlying RPC |  | ||||||
|         feeder `._rx_chan`. |  | ||||||
| 
 |  | ||||||
|         This knowledge is highly useful when handling so called |  | ||||||
|         "out-of-band" (OoB) cancellation conditions where a peer |  | ||||||
|         actor's task transmitted some remote error/cancel-msg and we |  | ||||||
|         must know whether to signal-via-cancel currently executing |  | ||||||
|         "user-code" (user defined code embedded in `ctx._scope`) or |  | ||||||
|         simply to forward the IPC-msg-as-error **without calling** |  | ||||||
|         `._scope.cancel()`. |  | ||||||
| 
 |  | ||||||
|         In the latter case it is presumed that if the owner task is |  | ||||||
|         blocking for the next IPC msg, it will eventually receive, |  | ||||||
|         process and raise the equivalent local error **without** |  | ||||||
|         requiring `._scope.cancel()` to be explicitly called by the |  | ||||||
|         *delivering OoB RPC-task* (via `_deliver_msg()`). |  | ||||||
| 
 |  | ||||||
|         ''' |  | ||||||
|         # NOTE, see the mem-chan meth-impls for *why* this |  | ||||||
|         # logic works, |  | ||||||
|         # `trio._channel.MemoryReceiveChannel.receive[_nowait]()` |  | ||||||
|         # |  | ||||||
|         # XXX realize that this is NOT an |  | ||||||
|         # official/will-be-loudly-deprecated API: |  | ||||||
|         # - https://trio.readthedocs.io/en/stable/reference-lowlevel.html#trio.lowlevel.Task.custom_sleep_data |  | ||||||
|         #  |_https://trio.readthedocs.io/en/stable/reference-lowlevel.html#trio.lowlevel.wait_task_rescheduled |  | ||||||
|         # |  | ||||||
|         # orig repo intro in the mem-chan change over patch: |  | ||||||
|         # - https://github.com/python-trio/trio/pull/586#issuecomment-414039117 |  | ||||||
|         #  |_https://github.com/python-trio/trio/pull/616 |  | ||||||
|         #  |_https://github.com/njsmith/trio/commit/98c38cef6f62e731bf8c7190e8756976bface8f0 |  | ||||||
|         # |  | ||||||
|         return ( |  | ||||||
|             self._task.custom_sleep_data |  | ||||||
|             is |  | ||||||
|             self._rx_chan |  | ||||||
|         ) |  | ||||||
| 
 |  | ||||||
|     def _maybe_cancel_and_set_remote_error( |     def _maybe_cancel_and_set_remote_error( | ||||||
|         self, |         self, | ||||||
|         error: BaseException, |         error: BaseException, | ||||||
|  | @ -852,27 +787,13 @@ class Context: | ||||||
|         if self._canceller is None: |         if self._canceller is None: | ||||||
|             log.error('Ctx has no canceller set!?') |             log.error('Ctx has no canceller set!?') | ||||||
| 
 | 
 | ||||||
|         cs: trio.CancelScope = self._scope |  | ||||||
| 
 |  | ||||||
|         # ?TODO? see comment @ .start_remote_task()` |  | ||||||
|         # |  | ||||||
|         # if not cs: |  | ||||||
|         #     from .devx import mk_pdb |  | ||||||
|         #     mk_pdb().set_trace() |  | ||||||
|         #     raise RuntimeError( |  | ||||||
|         #         f'IPC ctx was not be opened prior to remote error delivery !?\n' |  | ||||||
|         #         f'{self}\n' |  | ||||||
|         #         f'\n' |  | ||||||
|         #         f'`Portal.open_context()` must be entered (somewhere) beforehand!\n' |  | ||||||
|         #     ) |  | ||||||
| 
 |  | ||||||
|         # Cancel the local `._scope`, catch that |         # Cancel the local `._scope`, catch that | ||||||
|         # `._scope.cancelled_caught` and re-raise any remote error |         # `._scope.cancelled_caught` and re-raise any remote error | ||||||
|         # once exiting (or manually calling `.wait_for_result()`) the |         # once exiting (or manually calling `.wait_for_result()`) the | ||||||
|         # `.open_context()`  block. |         # `.open_context()`  block. | ||||||
|  |         cs: trio.CancelScope = self._scope | ||||||
|         if ( |         if ( | ||||||
|             cs |             cs | ||||||
|             and not cs.cancel_called |  | ||||||
| 
 | 
 | ||||||
|             # XXX this is an expected cancel request response |             # XXX this is an expected cancel request response | ||||||
|             # message and we **don't need to raise it** in the |             # message and we **don't need to raise it** in the | ||||||
|  | @ -881,7 +802,8 @@ class Context: | ||||||
|             # if `._cancel_called` then `.cancel_acked and .cancel_called` |             # if `._cancel_called` then `.cancel_acked and .cancel_called` | ||||||
|             # always should be set. |             # always should be set. | ||||||
|             and not self._is_self_cancelled() |             and not self._is_self_cancelled() | ||||||
|             # and not cs.cancelled_caught |             and not cs.cancel_called | ||||||
|  |             and not cs.cancelled_caught | ||||||
|         ): |         ): | ||||||
|             if ( |             if ( | ||||||
|                 msgerr |                 msgerr | ||||||
|  | @ -892,7 +814,7 @@ class Context: | ||||||
|                 not self._cancel_on_msgerr |                 not self._cancel_on_msgerr | ||||||
|             ): |             ): | ||||||
|                 message: str = ( |                 message: str = ( | ||||||
|                     f'NOT Cancelling `Context._scope` since,\n' |                     'NOT Cancelling `Context._scope` since,\n' | ||||||
|                     f'Context._cancel_on_msgerr = {self._cancel_on_msgerr}\n\n' |                     f'Context._cancel_on_msgerr = {self._cancel_on_msgerr}\n\n' | ||||||
|                     f'AND we got a msg-type-error!\n' |                     f'AND we got a msg-type-error!\n' | ||||||
|                     f'{error}\n' |                     f'{error}\n' | ||||||
|  | @ -902,43 +824,13 @@ class Context: | ||||||
|                 # `trio.Cancelled` subtype here ;) |                 # `trio.Cancelled` subtype here ;) | ||||||
|                 # https://github.com/goodboy/tractor/issues/368 |                 # https://github.com/goodboy/tractor/issues/368 | ||||||
|                 message: str = 'Cancelling `Context._scope` !\n\n' |                 message: str = 'Cancelling `Context._scope` !\n\n' | ||||||
|                 cs.cancel() |                 # from .devx import pause_from_sync | ||||||
| 
 |                 # pause_from_sync() | ||||||
|         # TODO, explicit condition for OoB (self-)cancellation? |                 self._scope.cancel() | ||||||
|         # - we called `Portal.cancel_actor()` from this actor |         else: | ||||||
|         #   and the peer ctx task delivered ctxc due to it. |             message: str = 'NOT cancelling `Context._scope` !\n\n' | ||||||
|         # - currently `self._is_self_cancelled()` will be true |  | ||||||
|         #   since the ctxc.canceller check will match us even though it |  | ||||||
|         #   wasn't from this ctx specifically! |  | ||||||
|         elif ( |  | ||||||
|             cs |  | ||||||
|             and self._is_self_cancelled() |  | ||||||
|             and not cs.cancel_called |  | ||||||
|         ): |  | ||||||
|             message: str = ( |  | ||||||
|                 'Cancelling `ctx._scope` due to OoB self-cancel ?!\n' |  | ||||||
|                 '\n' |  | ||||||
|             ) |  | ||||||
|             # from .devx import mk_pdb |             # from .devx import mk_pdb | ||||||
|             # mk_pdb().set_trace() |             # mk_pdb().set_trace() | ||||||
|             # TODO XXX, required to fix timeout failure in |  | ||||||
|             # `test_cancelled_lockacquire_in_ipctx_not_unmaskeed` |  | ||||||
|             # |  | ||||||
| 
 |  | ||||||
|             # XXX NOTE XXX, this is SUPER SUBTLE! |  | ||||||
|             # we only want to cancel our embedded `._scope` |  | ||||||
|             # if the ctx's current/using task is NOT blocked |  | ||||||
|             # on `._rx_chan.receive()` and on some other |  | ||||||
|             # `trio`-checkpoint since in the former case |  | ||||||
|             # any `._remote_error` will be relayed through |  | ||||||
|             # the rx-chan and appropriately raised by the owning |  | ||||||
|             # `._task` directly. IF the owner task is however |  | ||||||
|             # blocking elsewhere we need to interrupt it **now**. |  | ||||||
|             if not self._is_blocked_on_rx_chan(): |  | ||||||
|                 cs.cancel() |  | ||||||
|         else: |  | ||||||
|             # rx_stats = self._rx_chan.statistics() |  | ||||||
|             message: str = 'NOT cancelling `Context._scope` !\n\n' |  | ||||||
| 
 | 
 | ||||||
|         fmt_str: str = 'No `self._scope: CancelScope` was set/used ?\n' |         fmt_str: str = 'No `self._scope: CancelScope` was set/used ?\n' | ||||||
|         if ( |         if ( | ||||||
|  | @ -962,7 +854,6 @@ class Context: | ||||||
|                 + |                 + | ||||||
|                 cs_fmt |                 cs_fmt | ||||||
|             ) |             ) | ||||||
| 
 |  | ||||||
|         log.cancel( |         log.cancel( | ||||||
|             message |             message | ||||||
|             + |             + | ||||||
|  | @ -1055,9 +946,8 @@ class Context: | ||||||
| 
 | 
 | ||||||
|         ''' |         ''' | ||||||
|         side: str = self.side |         side: str = self.side | ||||||
|         self._cancel_called = True |         # XXX for debug via the `@.setter` | ||||||
|         # ^ XXX for debug via the `@.setter` |         self.cancel_called = True | ||||||
|         # self.cancel_called = True |  | ||||||
| 
 | 
 | ||||||
|         header: str = ( |         header: str = ( | ||||||
|             f'Cancelling ctx from {side!r}-side\n' |             f'Cancelling ctx from {side!r}-side\n' | ||||||
|  | @ -2121,9 +2011,6 @@ async def open_context_from_portal( | ||||||
|             f'|_{portal.actor}\n' |             f'|_{portal.actor}\n' | ||||||
|         ) |         ) | ||||||
| 
 | 
 | ||||||
|     # ?TODO? could we move this to inside the `tn` block? |  | ||||||
|     # -> would allow doing `ctx.parent_task = tn.parent_task` ? |  | ||||||
|     # -> would allow a `if not ._scope: => raise RTE` ? |  | ||||||
|     ctx: Context = await portal.actor.start_remote_task( |     ctx: Context = await portal.actor.start_remote_task( | ||||||
|         portal.channel, |         portal.channel, | ||||||
|         nsf=nsf, |         nsf=nsf, | ||||||
|  | @ -2150,7 +2037,6 @@ async def open_context_from_portal( | ||||||
|     scope_err: BaseException|None = None |     scope_err: BaseException|None = None | ||||||
|     ctxc_from_child: ContextCancelled|None = None |     ctxc_from_child: ContextCancelled|None = None | ||||||
|     try: |     try: | ||||||
|         # from .devx import pause |  | ||||||
|         async with ( |         async with ( | ||||||
|             collapse_eg(), |             collapse_eg(), | ||||||
|             trio.open_nursery() as tn, |             trio.open_nursery() as tn, | ||||||
|  | @ -2173,10 +2059,6 @@ async def open_context_from_portal( | ||||||
|             # the dialog, the `Error` msg should be raised from the `msg` |             # the dialog, the `Error` msg should be raised from the `msg` | ||||||
|             # handling block below. |             # handling block below. | ||||||
|             try: |             try: | ||||||
|                 log.runtime( |  | ||||||
|                     f'IPC ctx parent waiting on Started msg..\n' |  | ||||||
|                     f'ctx.cid: {ctx.cid!r}\n' |  | ||||||
|                 ) |  | ||||||
|                 started_msg, first = await ctx._pld_rx.recv_msg( |                 started_msg, first = await ctx._pld_rx.recv_msg( | ||||||
|                     ipc=ctx, |                     ipc=ctx, | ||||||
|                     expect_msg=Started, |                     expect_msg=Started, | ||||||
|  | @ -2185,16 +2067,16 @@ async def open_context_from_portal( | ||||||
|                 ) |                 ) | ||||||
|             except trio.Cancelled as taskc: |             except trio.Cancelled as taskc: | ||||||
|                 ctx_cs: trio.CancelScope = ctx._scope |                 ctx_cs: trio.CancelScope = ctx._scope | ||||||
|                 log.cancel( |  | ||||||
|                     f'IPC ctx was cancelled during "child" task sync due to\n\n' |  | ||||||
|                     f'.cid: {ctx.cid!r}\n' |  | ||||||
|                     f'.maybe_error: {ctx.maybe_error!r}\n' |  | ||||||
|                 ) |  | ||||||
|                 # await pause(shield=True) |  | ||||||
| 
 |  | ||||||
|                 if not ctx_cs.cancel_called: |                 if not ctx_cs.cancel_called: | ||||||
|                     raise |                     raise | ||||||
| 
 | 
 | ||||||
|  |                 # from .devx import pause | ||||||
|  |                 # await pause(shield=True) | ||||||
|  | 
 | ||||||
|  |                 log.cancel( | ||||||
|  |                     'IPC ctx was cancelled during "child" task sync due to\n\n' | ||||||
|  |                     f'{ctx.maybe_error}\n' | ||||||
|  |                 ) | ||||||
|                 # OW if the ctx's scope was cancelled manually, |                 # OW if the ctx's scope was cancelled manually, | ||||||
|                 # likely the `Context` was cancelled via a call to |                 # likely the `Context` was cancelled via a call to | ||||||
|                 # `._maybe_cancel_and_set_remote_error()` so ensure |                 # `._maybe_cancel_and_set_remote_error()` so ensure | ||||||
|  | @ -2390,16 +2272,13 @@ async def open_context_from_portal( | ||||||
|         match scope_err: |         match scope_err: | ||||||
|             case trio.Cancelled(): |             case trio.Cancelled(): | ||||||
|                 logmeth = log.cancel |                 logmeth = log.cancel | ||||||
|                 cause: str = 'cancelled' |  | ||||||
| 
 | 
 | ||||||
|             # XXX explicitly report on any non-graceful-taskc cases |             # XXX explicitly report on any non-graceful-taskc cases | ||||||
|             case _: |             case _: | ||||||
|                 cause: str = 'errored' |  | ||||||
|                 logmeth = log.exception |                 logmeth = log.exception | ||||||
| 
 | 
 | ||||||
|         logmeth( |         logmeth( | ||||||
|             f'ctx {ctx.side!r}-side {cause!r} with,\n' |             f'ctx {ctx.side!r}-side exited with {ctx.repr_outcome()!r}\n' | ||||||
|             f'{ctx.repr_outcome()!r}\n' |  | ||||||
|         ) |         ) | ||||||
| 
 | 
 | ||||||
|         if debug_mode(): |         if debug_mode(): | ||||||
|  | @ -2424,7 +2303,6 @@ async def open_context_from_portal( | ||||||
|         # told us it's cancelled ;p |         # told us it's cancelled ;p | ||||||
|         if ctxc_from_child is None: |         if ctxc_from_child is None: | ||||||
|             try: |             try: | ||||||
|                 # await pause(shield=True) |  | ||||||
|                 await ctx.cancel() |                 await ctx.cancel() | ||||||
|             except ( |             except ( | ||||||
|                 trio.BrokenResourceError, |                 trio.BrokenResourceError, | ||||||
|  | @ -2581,10 +2459,8 @@ async def open_context_from_portal( | ||||||
|                 log.cancel( |                 log.cancel( | ||||||
|                     f'Context cancelled by local {ctx.side!r}-side task\n' |                     f'Context cancelled by local {ctx.side!r}-side task\n' | ||||||
|                     f'c)>\n' |                     f'c)>\n' | ||||||
|                     f'  |_{ctx.parent_task}\n' |                     f' |_{ctx._task}\n\n' | ||||||
|                     f'   .cid={ctx.cid!r}\n' |                     f'{repr(scope_err)}\n' | ||||||
|                     f'\n' |  | ||||||
|                     f'{scope_err!r}\n' |  | ||||||
|                 ) |                 ) | ||||||
| 
 | 
 | ||||||
|             # TODO: should we add a `._cancel_req_received` |             # TODO: should we add a `._cancel_req_received` | ||||||
|  |  | ||||||
|  | @ -446,12 +446,12 @@ class ActorNursery: | ||||||
| @acm | @acm | ||||||
| async def _open_and_supervise_one_cancels_all_nursery( | async def _open_and_supervise_one_cancels_all_nursery( | ||||||
|     actor: Actor, |     actor: Actor, | ||||||
|     hide_tb: bool = True, |     tb_hide: bool = False, | ||||||
| 
 | 
 | ||||||
| ) -> typing.AsyncGenerator[ActorNursery, None]: | ) -> typing.AsyncGenerator[ActorNursery, None]: | ||||||
| 
 | 
 | ||||||
|     # normally don't need to show user by default |     # normally don't need to show user by default | ||||||
|     __tracebackhide__: bool = hide_tb |     __tracebackhide__: bool = tb_hide | ||||||
| 
 | 
 | ||||||
|     outer_err: BaseException|None = None |     outer_err: BaseException|None = None | ||||||
|     inner_err: BaseException|None = None |     inner_err: BaseException|None = None | ||||||
|  |  | ||||||
|  | @ -613,9 +613,10 @@ async def drain_to_final_msg( | ||||||
|             #       msg: dict = await ctx._rx_chan.receive() |             #       msg: dict = await ctx._rx_chan.receive() | ||||||
|             #   if res_cs.cancelled_caught: |             #   if res_cs.cancelled_caught: | ||||||
|             # |             # | ||||||
|             # -[x] make sure pause points work here for REPLing |             # -[ ] make sure pause points work here for REPLing | ||||||
|             #   the runtime itself; i.e. ensure there's no hangs! |             #   the runtime itself; i.e. ensure there's no hangs! | ||||||
|             #  |_see masked code below in .cancel_called path |             # |_from tractor.devx.debug import pause | ||||||
|  |             #   await pause() | ||||||
| 
 | 
 | ||||||
|         # NOTE: we get here if the far end was |         # NOTE: we get here if the far end was | ||||||
|         # `ContextCancelled` in 2 cases: |         # `ContextCancelled` in 2 cases: | ||||||
|  | @ -651,10 +652,6 @@ async def drain_to_final_msg( | ||||||
|                     f'IPC ctx cancelled externally during result drain ?\n' |                     f'IPC ctx cancelled externally during result drain ?\n' | ||||||
|                     f'{ctx}' |                     f'{ctx}' | ||||||
|                 ) |                 ) | ||||||
|                 # XXX, for tracing `Cancelled`.. |  | ||||||
|                 # from tractor.devx.debug import pause |  | ||||||
|                 # await pause(shield=True) |  | ||||||
| 
 |  | ||||||
|             # CASE 2: mask the local cancelled-error(s) |             # CASE 2: mask the local cancelled-error(s) | ||||||
|             # only when we are sure the remote error is |             # only when we are sure the remote error is | ||||||
|             # the source cause of this local task's |             # the source cause of this local task's | ||||||
|  |  | ||||||
|  | @ -72,7 +72,7 @@ _mask_cases: dict[ | ||||||
|     dict[ |     dict[ | ||||||
|         int,  # inner-frame index into `inspect.getinnerframes()` |         int,  # inner-frame index into `inspect.getinnerframes()` | ||||||
|         # `FrameInfo.function/filename: str`s to match |         # `FrameInfo.function/filename: str`s to match | ||||||
|         dict[str, str], |         tuple[str, str], | ||||||
|     ], |     ], | ||||||
| ] = { | ] = { | ||||||
|     trio.WouldBlock: { |     trio.WouldBlock: { | ||||||
|  | @ -275,8 +275,7 @@ async def maybe_raise_from_masking_exc( | ||||||
|                         )) |                         )) | ||||||
|                     ): |                     ): | ||||||
|                         log.warning( |                         log.warning( | ||||||
|                             f'Ignoring already-known, non-ideal-but-valid ' |                             f'Ignoring already-known/non-ideally-valid masker code @\n' | ||||||
|                             f'masker code @\n' |  | ||||||
|                             f'{masker_frame}\n' |                             f'{masker_frame}\n' | ||||||
|                             f'\n' |                             f'\n' | ||||||
|                             f'NOT raising {exc_ctx} from masker {exc_match!r}\n' |                             f'NOT raising {exc_ctx} from masker {exc_match!r}\n' | ||||||
|  |  | ||||||
		Loading…
	
		Reference in New Issue