Only `ctx.canccel()` when non-callee ctxc received
Since presumably if the ctxc *is* received from the callee side, there's no point in also sending it a cancel request ;) Also, try out shielding the `ctx.cancel()` call inside `Portal.open_context()` since that seems to resolve testing assertions where you'd expect `.cancel_called` to have been set in the case where an inter-peer cancellation is raised by some local ctx-handle and that local ctxc causes other surrounding `.open_context()` blocks to raise and then implicitly call `ctx.cancel()`. However, there's a possible race where the callee side might deliver a ctxc while all that is happening, `._scope.cancel()` gets called and then by the time `ctx.cancel()` is called it raises `trio.Cancelled` instead continuing to set `._cancel_called`? This needs more testing and docs to get the minute details correct .. yet again.shielded_ctx_cancel
parent
9e3f41a5b1
commit
0e310971a3
|
@ -42,7 +42,7 @@ from async_generator import asynccontextmanager
|
||||||
from .trionics import maybe_open_nursery
|
from .trionics import maybe_open_nursery
|
||||||
from .devx import (
|
from .devx import (
|
||||||
# acquire_debug_lock,
|
# acquire_debug_lock,
|
||||||
# pause,
|
pause,
|
||||||
maybe_wait_for_debugger,
|
maybe_wait_for_debugger,
|
||||||
)
|
)
|
||||||
from ._state import (
|
from ._state import (
|
||||||
|
@ -556,7 +556,7 @@ class Portal:
|
||||||
# placeholder for any exception raised in the runtime
|
# placeholder for any exception raised in the runtime
|
||||||
# or by user tasks which cause this context's closure.
|
# or by user tasks which cause this context's closure.
|
||||||
scope_err: BaseException|None = None
|
scope_err: BaseException|None = None
|
||||||
ctxc_from_callee: ContextCancelled|None = None
|
ctxc_from_callee: ContextCancelled|bool = False
|
||||||
try:
|
try:
|
||||||
async with trio.open_nursery() as nurse:
|
async with trio.open_nursery() as nurse:
|
||||||
|
|
||||||
|
@ -673,8 +673,6 @@ class Portal:
|
||||||
# `Nursery.cancel_scope.cancel()`)
|
# `Nursery.cancel_scope.cancel()`)
|
||||||
except ContextCancelled as ctxc:
|
except ContextCancelled as ctxc:
|
||||||
scope_err = ctxc
|
scope_err = ctxc
|
||||||
ctxc_from_callee = ctxc
|
|
||||||
|
|
||||||
# XXX TODO XXX: FIX THIS debug_mode BUGGGG!!!
|
# XXX TODO XXX: FIX THIS debug_mode BUGGGG!!!
|
||||||
# using this code and then resuming the REPL will
|
# using this code and then resuming the REPL will
|
||||||
# cause a SIGINT-ignoring HANG!
|
# cause a SIGINT-ignoring HANG!
|
||||||
|
@ -686,6 +684,10 @@ class Portal:
|
||||||
#
|
#
|
||||||
# await pause()
|
# await pause()
|
||||||
|
|
||||||
|
# ctxc_from_callee: bool = (
|
||||||
|
# ctxc.src_actor_uid == self.chan.uid
|
||||||
|
# )
|
||||||
|
|
||||||
# CASE 2: context was cancelled by local task calling
|
# CASE 2: context was cancelled by local task calling
|
||||||
# `.cancel()`, we don't raise and the exit block should
|
# `.cancel()`, we don't raise and the exit block should
|
||||||
# exit silently.
|
# exit silently.
|
||||||
|
@ -696,6 +698,7 @@ class Portal:
|
||||||
and
|
and
|
||||||
ctxc.canceller == self.actor.uid
|
ctxc.canceller == self.actor.uid
|
||||||
):
|
):
|
||||||
|
ctxc_from_callee = True
|
||||||
log.cancel(
|
log.cancel(
|
||||||
f'Context (cid=[{ctx.cid[-6:]}..] cancelled gracefully with:\n'
|
f'Context (cid=[{ctx.cid[-6:]}..] cancelled gracefully with:\n'
|
||||||
f'{ctxc}'
|
f'{ctxc}'
|
||||||
|
@ -755,6 +758,7 @@ class Portal:
|
||||||
f'{caller_err}\n'
|
f'{caller_err}\n'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# TODO: does this need to be shielded?
|
||||||
if debug_mode():
|
if debug_mode():
|
||||||
# async with acquire_debug_lock(self.actor.uid):
|
# async with acquire_debug_lock(self.actor.uid):
|
||||||
# pass
|
# pass
|
||||||
|
@ -771,12 +775,14 @@ class Portal:
|
||||||
'Calling `ctx.cancel()`!\n'
|
'Calling `ctx.cancel()`!\n'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
# we don't need to cancel the callee if it already
|
# we don't need to cancel the callee if it already
|
||||||
# told us it's cancelled ;p
|
# told us it's cancelled ;p
|
||||||
if ctxc_from_callee is None:
|
if not ctxc_from_callee:
|
||||||
|
log.critical('SENDING CANCEL')
|
||||||
try:
|
try:
|
||||||
await ctx.cancel()
|
# await pause(shield=True)
|
||||||
|
with trio.CancelScope(shield=True):
|
||||||
|
await ctx.cancel()
|
||||||
except (
|
except (
|
||||||
trio.BrokenResourceError,
|
trio.BrokenResourceError,
|
||||||
trio.ClosedResourceError,
|
trio.ClosedResourceError,
|
||||||
|
|
Loading…
Reference in New Issue