forked from goodboy/tractor
commit
0f7db27b68
|
@ -0,0 +1,7 @@
|
||||||
|
Drop `trio.Process.aclose()` usage, copy into our spawning code.
|
||||||
|
|
||||||
|
The details are laid out in https://github.com/goodboy/tractor/issues/330.
|
||||||
|
`trio` changed is process running quite some time ago, this just copies
|
||||||
|
out the small bit we needed (from the old `.aclose()`) for hard kills
|
||||||
|
where a soft runtime cancel request fails and our "zombie killer"
|
||||||
|
implementation kicks in.
|
|
@ -23,13 +23,12 @@ import sys
|
||||||
import platform
|
import platform
|
||||||
from typing import (
|
from typing import (
|
||||||
Any,
|
Any,
|
||||||
|
Awaitable,
|
||||||
Literal,
|
Literal,
|
||||||
Optional,
|
|
||||||
Callable,
|
Callable,
|
||||||
TypeVar,
|
TypeVar,
|
||||||
TYPE_CHECKING,
|
TYPE_CHECKING,
|
||||||
)
|
)
|
||||||
from collections.abc import Awaitable
|
|
||||||
|
|
||||||
from exceptiongroup import BaseExceptionGroup
|
from exceptiongroup import BaseExceptionGroup
|
||||||
import trio
|
import trio
|
||||||
|
@ -60,7 +59,7 @@ if TYPE_CHECKING:
|
||||||
log = get_logger('tractor')
|
log = get_logger('tractor')
|
||||||
|
|
||||||
# placeholder for an mp start context if so using that backend
|
# placeholder for an mp start context if so using that backend
|
||||||
_ctx: Optional[mp.context.BaseContext] = None
|
_ctx: mp.context.BaseContext | None = None
|
||||||
SpawnMethodKey = Literal[
|
SpawnMethodKey = Literal[
|
||||||
'trio', # supported on all platforms
|
'trio', # supported on all platforms
|
||||||
'mp_spawn',
|
'mp_spawn',
|
||||||
|
@ -86,7 +85,7 @@ else:
|
||||||
def try_set_start_method(
|
def try_set_start_method(
|
||||||
key: SpawnMethodKey
|
key: SpawnMethodKey
|
||||||
|
|
||||||
) -> Optional[mp.context.BaseContext]:
|
) -> mp.context.BaseContext | None:
|
||||||
'''
|
'''
|
||||||
Attempt to set the method for process starting, aka the "actor
|
Attempt to set the method for process starting, aka the "actor
|
||||||
spawning backend".
|
spawning backend".
|
||||||
|
@ -200,16 +199,37 @@ async def cancel_on_completion(
|
||||||
async def do_hard_kill(
|
async def do_hard_kill(
|
||||||
proc: trio.Process,
|
proc: trio.Process,
|
||||||
terminate_after: int = 3,
|
terminate_after: int = 3,
|
||||||
|
|
||||||
) -> None:
|
) -> None:
|
||||||
# NOTE: this timeout used to do nothing since we were shielding
|
# NOTE: this timeout used to do nothing since we were shielding
|
||||||
# the ``.wait()`` inside ``new_proc()`` which will pretty much
|
# the ``.wait()`` inside ``new_proc()`` which will pretty much
|
||||||
# never release until the process exits, now it acts as
|
# never release until the process exits, now it acts as
|
||||||
# a hard-kill time ultimatum.
|
# a hard-kill time ultimatum.
|
||||||
|
log.debug(f"Terminating {proc}")
|
||||||
with trio.move_on_after(terminate_after) as cs:
|
with trio.move_on_after(terminate_after) as cs:
|
||||||
|
|
||||||
# NOTE: This ``__aexit__()`` shields internally.
|
# NOTE: code below was copied verbatim from the now deprecated
|
||||||
async with proc: # calls ``trio.Process.aclose()``
|
# (in 0.20.0) ``trio._subrocess.Process.aclose()``, orig doc
|
||||||
log.debug(f"Terminating {proc}")
|
# string:
|
||||||
|
#
|
||||||
|
# Close any pipes we have to the process (both input and output)
|
||||||
|
# and wait for it to exit. If cancelled, kills the process and
|
||||||
|
# waits for it to finish exiting before propagating the
|
||||||
|
# cancellation.
|
||||||
|
with trio.CancelScope(shield=True):
|
||||||
|
if proc.stdin is not None:
|
||||||
|
await proc.stdin.aclose()
|
||||||
|
if proc.stdout is not None:
|
||||||
|
await proc.stdout.aclose()
|
||||||
|
if proc.stderr is not None:
|
||||||
|
await proc.stderr.aclose()
|
||||||
|
try:
|
||||||
|
await proc.wait()
|
||||||
|
finally:
|
||||||
|
if proc.returncode is None:
|
||||||
|
proc.kill()
|
||||||
|
with trio.CancelScope(shield=True):
|
||||||
|
await proc.wait()
|
||||||
|
|
||||||
if cs.cancelled_caught:
|
if cs.cancelled_caught:
|
||||||
# XXX: should pretty much never get here unless we have
|
# XXX: should pretty much never get here unless we have
|
||||||
|
@ -355,12 +375,11 @@ async def trio_proc(
|
||||||
spawn_cmd.append("--asyncio")
|
spawn_cmd.append("--asyncio")
|
||||||
|
|
||||||
cancelled_during_spawn: bool = False
|
cancelled_during_spawn: bool = False
|
||||||
proc: Optional[trio.Process] = None
|
proc: trio.Process | None = None
|
||||||
try:
|
try:
|
||||||
try:
|
try:
|
||||||
# TODO: needs ``trio_typing`` patch?
|
# TODO: needs ``trio_typing`` patch?
|
||||||
proc = await trio.lowlevel.open_process( # type: ignore
|
proc = await trio.lowlevel.open_process(spawn_cmd)
|
||||||
spawn_cmd)
|
|
||||||
|
|
||||||
log.runtime(f"Started {proc}")
|
log.runtime(f"Started {proc}")
|
||||||
|
|
||||||
|
@ -444,8 +463,8 @@ async def trio_proc(
|
||||||
nursery.cancel_scope.cancel()
|
nursery.cancel_scope.cancel()
|
||||||
|
|
||||||
finally:
|
finally:
|
||||||
# The "hard" reap since no actor zombies are allowed!
|
# XXX NOTE XXX: The "hard" reap since no actor zombies are
|
||||||
# XXX: do this **after** cancellation/tearfown to avoid
|
# allowed! Do this **after** cancellation/teardown to avoid
|
||||||
# killing the process too early.
|
# killing the process too early.
|
||||||
if proc:
|
if proc:
|
||||||
log.cancel(f'Hard reap sequence starting for {subactor.uid}')
|
log.cancel(f'Hard reap sequence starting for {subactor.uid}')
|
||||||
|
|
Loading…
Reference in New Issue