Compare commits
1 Commits
203a0f7e1f
...
ee875c930b
| Author | SHA1 | Date |
|---|---|---|
|
|
ee875c930b |
|
|
@ -1,39 +0,0 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
session: moc-teardown-completion-20260804
|
||||
timestamp: 2026-08-04T03:03:09Z
|
||||
git_ref: 65bf9df5
|
||||
scope: code
|
||||
substantive: true
|
||||
raw_file: 20260804T030309Z_65bf9df5_prompt_io.raw.md
|
||||
---
|
||||
|
||||
## Prompt
|
||||
|
||||
Patch Tractor's `maybe_open_context()` so the final consumer waits for
|
||||
resource `__aexit__()` completion and receives cleanup errors. Reuse
|
||||
`outcome.Outcome` for the exit result; use the smaller mutable
|
||||
`_CtxExit` holder if that makes the implementation simpler. Add and run
|
||||
the relevant existing unit tests, but do not commit or push the patch.
|
||||
After reviewing the result, simplify `_CtxExit` back to an optional
|
||||
exception because the success outcome carries no useful value.
|
||||
|
||||
## Response summary
|
||||
|
||||
Added an exception-backed completion handshake between
|
||||
`_Cache.run_ctx()` and the final `maybe_open_context()` consumer.
|
||||
Serialized consumer registration and final teardown under the per-key
|
||||
lock, preserving that lock for queued entrants. Added deterministic
|
||||
regressions for normal exit, cleanup errors, cancellation interactions,
|
||||
service-nursery cancellation, and teardown re-entry.
|
||||
|
||||
## Files changed
|
||||
|
||||
- `tractor/trionics/_mngrs.py` - publish and unwrap cached exit outcomes.
|
||||
- `tests/test_resource_cache.py` - cover completion and cancellation.
|
||||
|
||||
## Human edits
|
||||
|
||||
The user directed the final simplification from `outcome.Outcome` to an
|
||||
optional exception field. The patch remains uncommitted.
|
||||
|
|
@ -1,36 +0,0 @@
|
|||
---
|
||||
model: openai/gpt-5.6-sol
|
||||
service: opencode
|
||||
timestamp: 2026-08-04T03:03:09Z
|
||||
git_ref: 65bf9df5
|
||||
diff_cmd: git diff HEAD~1..HEAD
|
||||
---
|
||||
|
||||
Implemented cached-context exit completion in
|
||||
`tractor.trionics.maybe_open_context()`.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tractor/trionics/_mngrs.py`
|
||||
|
||||
The generated implementation adds `_CtxExit`, whose `done` event
|
||||
publishes an `outcome.Outcome[None]`. `_Cache.run_ctx()` records either
|
||||
`Value(None)` or `Error(exc)` after the resource exit attempt. The final
|
||||
MOC consumer signals `no_more_users`, waits for completion under a
|
||||
shielded cancel scope, removes the per-key lock, and unwraps the outcome
|
||||
so ordinary cleanup failures are raised at the consumer boundary.
|
||||
`trio.Cancelled`, `KeyboardInterrupt`, and `SystemExit` continue through
|
||||
the service task rather than being converted into regular cleanup
|
||||
errors.
|
||||
|
||||
> `git diff HEAD~1..HEAD -- tests/test_resource_cache.py`
|
||||
|
||||
The generated regressions cover successful exit blocking, cleanup-error
|
||||
delivery, final-user cancellation, cancellation combined with a cleanup
|
||||
error, and service-nursery cancellation. The existing teardown re-entry
|
||||
test now uses explicit events instead of a ten-second cleanup sleep and
|
||||
asserts that the replacement resource is a fresh cache miss.
|
||||
|
||||
Verification:
|
||||
|
||||
`env PYTHONPATH="$PWD" /home/goodboy/repos/tractor/py313/bin/python -m pytest tests/test_resource_cache.py`
|
||||
|
||||
Result: `16 passed in 8.37s`.
|
||||
|
|
@ -1,22 +0,0 @@
|
|||
# AI Prompt I/O Log - OpenCode
|
||||
|
||||
This directory tracks prompt inputs and model outputs for AI-assisted
|
||||
development using `opencode`.
|
||||
|
||||
## Policy
|
||||
|
||||
Prompt logging follows the [NLNet generative AI policy][nlnet-ai]. All
|
||||
substantive AI contributions are logged with:
|
||||
|
||||
- Model name and version
|
||||
- Timestamps
|
||||
- The prompts that produced the output
|
||||
- Unedited model output (`.raw.md` files)
|
||||
|
||||
[nlnet-ai]: https://nlnet.nl/foundation/policies/generativeAI/
|
||||
|
||||
## Usage
|
||||
|
||||
Entries are created by the prompt-io workflow. Human contributors remain
|
||||
accountable for all decisions. AI-generated content is never presented as
|
||||
human-authored work.
|
||||
|
|
@ -2,19 +2,16 @@
|
|||
`tractor.log`-wrapping unit tests.
|
||||
|
||||
'''
|
||||
import logging
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
from types import ModuleType
|
||||
|
||||
import pytest
|
||||
import tractor
|
||||
import trio
|
||||
from tractor import (
|
||||
_code_load,
|
||||
log,
|
||||
)
|
||||
from tractor.ipc import _chan
|
||||
|
||||
|
||||
def test_root_pkg_not_duplicated_in_logger_name():
|
||||
|
|
@ -225,88 +222,6 @@ def test_add_log_level_pluggable():
|
|||
delattr(log.StackLevelAdapter, name.lower())
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'suppression',
|
||||
[
|
||||
'level',
|
||||
'logger',
|
||||
'global',
|
||||
],
|
||||
)
|
||||
def test_log_guard_skips_payload_formatting(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
suppression: str,
|
||||
):
|
||||
'''
|
||||
Suppressed transport logs must not render payloads.
|
||||
|
||||
The original hot-path guard compared only the effective logger
|
||||
level. A logger disabled through its `Logger.disabled` flag or
|
||||
the global `logging.disable()` threshold could therefore still
|
||||
call `pformat()` before `Logger.isEnabledFor()` discarded the
|
||||
record.
|
||||
|
||||
Exercise effective-level, per-logger, and global suppression
|
||||
independently. A poisoned `_chan.pformat()` proves rendering is
|
||||
skipped, while the fake transport proves `Channel.send()` still
|
||||
transmits the original payload and traceback-hiding flag.
|
||||
|
||||
'''
|
||||
sent: list[tuple[object, bool]] = []
|
||||
|
||||
class FakeTransport:
|
||||
async def send(
|
||||
self,
|
||||
payload: object,
|
||||
hide_tb: bool = False,
|
||||
) -> None:
|
||||
sent.append((payload, hide_tb))
|
||||
|
||||
def fail_pformat(payload: object) -> str:
|
||||
raise AssertionError(
|
||||
f'suppressed log rendered payload: {payload!r}'
|
||||
)
|
||||
|
||||
chan_log = log.get_logger(
|
||||
name=f'guard_test.{suppression}',
|
||||
)
|
||||
std_log = chan_log.logger
|
||||
orig_level: int = std_log.level
|
||||
orig_disable: int = logging.root.manager.disable
|
||||
transport_level: int = log.CUSTOM_LEVELS['TRANSPORT']
|
||||
|
||||
monkeypatch.setattr(_chan, 'log', chan_log)
|
||||
monkeypatch.setattr(_chan, 'pformat', fail_pformat)
|
||||
try:
|
||||
logging.disable(logging.NOTSET)
|
||||
std_log.setLevel(transport_level)
|
||||
|
||||
if suppression == 'level':
|
||||
std_log.setLevel(logging.INFO)
|
||||
elif suppression == 'logger':
|
||||
monkeypatch.setattr(std_log, 'disabled', True)
|
||||
else:
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
assert not chan_log.isEnabledFor(transport_level)
|
||||
|
||||
transport = FakeTransport()
|
||||
chan = _chan.Channel(transport=transport)
|
||||
payload = object()
|
||||
|
||||
async def send_payload() -> None:
|
||||
await chan.send(
|
||||
payload,
|
||||
hide_tb=True,
|
||||
)
|
||||
|
||||
trio.run(send_payload)
|
||||
assert sent == [(payload, True)]
|
||||
finally:
|
||||
std_log.setLevel(orig_level)
|
||||
logging.disable(orig_disable)
|
||||
|
||||
|
||||
# TODO, moar tests against existing feats:
|
||||
# ------ - ------
|
||||
# - [ ] color settings?
|
||||
|
|
|
|||
|
|
@ -9,7 +9,6 @@ from typing import Awaitable
|
|||
|
||||
import pytest
|
||||
import trio
|
||||
from trio.testing import wait_all_tasks_blocked
|
||||
import tractor
|
||||
from tractor.trionics import (
|
||||
maybe_open_context,
|
||||
|
|
@ -95,232 +94,6 @@ def test_resource_only_entered_once(key_on):
|
|||
trio.run(main)
|
||||
|
||||
|
||||
def test_last_moc_user_waits_for_resource_exit():
|
||||
'''
|
||||
Verify the final user cannot return before resource teardown.
|
||||
|
||||
Previously the final `maybe_open_context()` user only signalled
|
||||
`_Cache.run_ctx()` through its `no_more_users` event. The user
|
||||
then returned while the service task was still running the
|
||||
resource's `__aexit__()`, so callers could observe stale external
|
||||
state immediately after their `async with` block.
|
||||
|
||||
The resource sets `exit_started` before blocking on
|
||||
`allow_exit`. The user task must remain inside MOC until the test
|
||||
releases that deterministic checkpoint and `__aexit__()` sets
|
||||
`exit_finished`.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
exit_started = trio.Event()
|
||||
allow_exit = trio.Event()
|
||||
exit_finished = trio.Event()
|
||||
user_returned = trio.Event()
|
||||
|
||||
@acm
|
||||
async def open_resource():
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
exit_started.set()
|
||||
await allow_exit.wait()
|
||||
exit_finished.set()
|
||||
|
||||
async def use_resource():
|
||||
async with maybe_open_context(open_resource):
|
||||
pass
|
||||
|
||||
assert exit_finished.is_set()
|
||||
user_returned.set()
|
||||
|
||||
async with (
|
||||
tractor.open_root_actor(),
|
||||
trio.open_nursery() as tn,
|
||||
):
|
||||
tn.start_soon(use_resource)
|
||||
await exit_started.wait()
|
||||
assert not user_returned.is_set()
|
||||
allow_exit.set()
|
||||
await user_returned.wait()
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_moc_delivers_resource_exit_error():
|
||||
'''
|
||||
Verify a resource exit error reaches the final MOC user.
|
||||
|
||||
Previously `_Cache.run_ctx()` executed the cached resource's
|
||||
`__aexit__()` after the final user had returned. An exit failure
|
||||
therefore surfaced later through the actor service nursery rather
|
||||
than at the user's `async with maybe_open_context()` boundary.
|
||||
|
||||
This resource raises a unique `ResourceExitError` during exit.
|
||||
Catching that exact instance around MOC proves the service task
|
||||
delivered the failure to the final user without replacing it.
|
||||
|
||||
'''
|
||||
class ResourceExitError(Exception):
|
||||
pass
|
||||
|
||||
exit_error = ResourceExitError('resource exit failed')
|
||||
|
||||
async def main():
|
||||
@acm
|
||||
async def open_resource():
|
||||
yield
|
||||
raise exit_error
|
||||
|
||||
async with tractor.open_root_actor():
|
||||
with pytest.raises(ResourceExitError) as exc_info:
|
||||
async with maybe_open_context(open_resource):
|
||||
pass
|
||||
|
||||
assert exc_info.value is exit_error
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_moc_final_user_cancellation_waits_for_exit():
|
||||
'''
|
||||
Verify final-user cancellation still waits for successful exit.
|
||||
|
||||
Previously cancellation escaped the final MOC user immediately
|
||||
after it signalled `_Cache.run_ctx()`, leaving resource exit to
|
||||
finish later in the actor service task. This violated the context
|
||||
manager boundary even when cleanup itself succeeded.
|
||||
|
||||
The consumer cancels its own scope while holding the sole cached
|
||||
resource. The resource sets `exit_finished` from its `finally`
|
||||
block, and the consumer checks that event immediately after its
|
||||
cancel scope catches `trio.Cancelled`. This proves MOC's
|
||||
completion wait is shielded without suppressing the original
|
||||
cancellation.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
exit_finished = trio.Event()
|
||||
|
||||
@acm
|
||||
async def open_resource():
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
exit_finished.set()
|
||||
|
||||
async with tractor.open_root_actor():
|
||||
with trio.CancelScope() as cs:
|
||||
async with maybe_open_context(open_resource):
|
||||
cs.cancel()
|
||||
await trio.sleep_forever()
|
||||
|
||||
assert cs.cancelled_caught
|
||||
assert exit_finished.is_set()
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_moc_exit_error_masks_final_user_cancellation():
|
||||
'''
|
||||
Verify cleanup errors survive final-user cancellation.
|
||||
|
||||
A cancelled final user previously signalled `no_more_users` and
|
||||
propagated `trio.Cancelled` before `_Cache.run_ctx()` completed
|
||||
resource exit. If `__aexit__()` then failed, its error was
|
||||
detached from the API call which caused teardown.
|
||||
|
||||
The consumer cancels its own scope at a deterministic checkpoint
|
||||
inside MOC. Resource exit raises `ResourceExitError`; observing
|
||||
that exact error outside the cancel scope proves MOC shields the
|
||||
completion wait and applies normal context-manager masking, where
|
||||
a cleanup failure replaces the active cancellation.
|
||||
|
||||
'''
|
||||
class ResourceExitError(Exception):
|
||||
pass
|
||||
|
||||
exit_error = ResourceExitError('resource exit failed')
|
||||
|
||||
async def main():
|
||||
@acm
|
||||
async def open_resource():
|
||||
yield
|
||||
raise exit_error
|
||||
|
||||
async with tractor.open_root_actor():
|
||||
with pytest.raises(ResourceExitError) as exc_info:
|
||||
with trio.CancelScope() as cs:
|
||||
async with maybe_open_context(open_resource):
|
||||
cs.cancel()
|
||||
await trio.sleep_forever()
|
||||
|
||||
assert exc_info.value is exit_error
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
def test_moc_service_nursery_cancellation_completes_exit():
|
||||
'''
|
||||
Verify service-nursery cancellation cannot strand a final user.
|
||||
|
||||
`_Cache.run_ctx()` and an MOC consumer may share a
|
||||
caller-provided service nursery. Cancelling that nursery
|
||||
interrupts the service task's `no_more_users` wait and the
|
||||
consumer body together. A shielded final-user wait would deadlock
|
||||
if `run_ctx()` failed to publish completion while propagating its
|
||||
own `trio.Cancelled`.
|
||||
|
||||
The outer task waits for resource entry, then cancels the exact
|
||||
nursery containing both tasks. The resource shields one cleanup
|
||||
checkpoint and sets `exit_finished`; observing both it and
|
||||
`service_finished` proves cancellation propagated normally while
|
||||
MOC's completion handshake terminated deterministically.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
resource_entered = trio.Event()
|
||||
exit_finished = trio.Event()
|
||||
service_finished = trio.Event()
|
||||
service_tn: trio.Nursery|None = None
|
||||
|
||||
@acm
|
||||
async def open_resource():
|
||||
try:
|
||||
resource_entered.set()
|
||||
yield
|
||||
finally:
|
||||
with trio.CancelScope(shield=True):
|
||||
await trio.lowlevel.checkpoint()
|
||||
exit_finished.set()
|
||||
|
||||
async def use_resource(tn: trio.Nursery):
|
||||
async with maybe_open_context(
|
||||
open_resource,
|
||||
tn=tn,
|
||||
):
|
||||
await trio.sleep_forever()
|
||||
|
||||
async def run_service():
|
||||
nonlocal service_tn
|
||||
|
||||
async with trio.open_nursery() as tn:
|
||||
service_tn = tn
|
||||
tn.start_soon(use_resource, tn)
|
||||
|
||||
service_finished.set()
|
||||
|
||||
async with trio.open_nursery() as outer_tn:
|
||||
outer_tn.start_soon(run_service)
|
||||
await resource_entered.wait()
|
||||
assert service_tn is not None
|
||||
service_tn.cancel_scope.cancel()
|
||||
await service_finished.wait()
|
||||
|
||||
assert exit_finished.is_set()
|
||||
|
||||
trio.run(main)
|
||||
|
||||
|
||||
@tractor.context
|
||||
async def streamer(
|
||||
ctx: tractor.Context,
|
||||
|
|
@ -775,52 +548,43 @@ def test_moc_reentry_during_teardown(
|
|||
loglevel: str,
|
||||
):
|
||||
'''
|
||||
Reproduce re-entry while an identical cached context exits.
|
||||
Reproduce the piker `open_cached_client('kraken')` race:
|
||||
|
||||
- multiple tasks use the same `acm_func` with no kwargs,
|
||||
producing an identical `ctx_key`;
|
||||
- all users leave and the final user starts resource teardown;
|
||||
- `_Cache.run_ctx()` removes the cached value and resource entry
|
||||
before entering the resource's blocking `__aexit__()` body;
|
||||
- a new task attempts to enter that same `ctx_key` during exit;
|
||||
- the per-key lock keeps that entrant queued until exit
|
||||
completes;
|
||||
- the entrant then receives a fresh cache miss and resource.
|
||||
- same `acm_func`, NO kwargs (identical `ctx_key`)
|
||||
- multiple tasks share the cached resource
|
||||
- all users exit -> teardown starts
|
||||
- a NEW task enters during `_Cache.run_ctx.__aexit__`
|
||||
- `values[ctx_key]` is gone (popped in inner finally)
|
||||
but `resources[ctx_key]` still exists (outer finally
|
||||
hasn't run yet bc the acm cleanup has checkpoints)
|
||||
- old code: `assert not resources.get(ctx_key)` FIRES
|
||||
|
||||
Without teardown sharing the registration lock, re-entry could
|
||||
race resource replacement while the prior generation was still
|
||||
exiting. The final user could also return before that exit
|
||||
completed.
|
||||
|
||||
The first resource generation signals `in_aexit` and waits on
|
||||
`allow_aexit`. The re-entry task signals `reentry_started` and
|
||||
blocks inside MOC; only after `wait_all_tasks_blocked()` confirms
|
||||
that ordering does the coordinator release cleanup. The entrant
|
||||
must then receive a fresh cache miss. `first_done` additionally
|
||||
proves the first MOC user observed completed teardown before
|
||||
returning.
|
||||
This models the real-world scenario where `brokerd.kraken`
|
||||
tasks concurrently call `open_cached_client('kraken')`
|
||||
(same `acm_func`, empty kwargs, shared `ctx_key`) and
|
||||
the teardown/re-entry race triggers intermittently.
|
||||
|
||||
'''
|
||||
async def main():
|
||||
in_aexit = trio.Event()
|
||||
allow_aexit = trio.Event()
|
||||
reentry_started = trio.Event()
|
||||
generation: int = 0
|
||||
|
||||
@acm
|
||||
async def cached_client():
|
||||
'''
|
||||
Simulate a no-argument `kraken.api.get_client()`.
|
||||
Simulates `kraken.api.get_client()`:
|
||||
- no params (all callers share one `ctx_key`)
|
||||
- slow-ish cleanup to widen the race window
|
||||
between `values.pop()` and `resources.pop()`
|
||||
inside `_Cache.run_ctx`.
|
||||
|
||||
'''
|
||||
nonlocal generation
|
||||
|
||||
generation += 1
|
||||
resource_generation: int = generation
|
||||
yield 'the-client'
|
||||
if resource_generation == 1:
|
||||
# Signal that we're in __aexit__ — at this
|
||||
# point `values` has already been popped by
|
||||
# `run_ctx`'s inner finally, but `resources`
|
||||
# is still alive (outer finally hasn't run).
|
||||
in_aexit.set()
|
||||
await allow_aexit.wait()
|
||||
await trio.sleep(10)
|
||||
|
||||
first_done = trio.Event()
|
||||
|
||||
|
|
@ -834,25 +598,16 @@ def test_moc_reentry_during_teardown(
|
|||
async def reenter_during_teardown():
|
||||
'''
|
||||
Wait for the acm's `__aexit__` to start (meaning
|
||||
the cached value is no longer available), then re-enter.
|
||||
`values` is popped but `resources` still exists),
|
||||
then re-enter — triggering the assert.
|
||||
|
||||
'''
|
||||
await in_aexit.wait()
|
||||
|
||||
# Tell the coordinator this task is about to enter MOC.
|
||||
# `Event.set()` is not a checkpoint. Though `async with`
|
||||
# awaits MOC's `__aenter__()`, its async generator runs
|
||||
# synchronously until the held per-key `lock.acquire()`
|
||||
# actually suspends this task.
|
||||
reentry_started.set()
|
||||
async with maybe_open_context(
|
||||
cached_client,
|
||||
) as (cache_hit, value):
|
||||
assert not cache_hit
|
||||
assert value == 'the-client'
|
||||
|
||||
await first_done.wait()
|
||||
|
||||
with trio.fail_after(5):
|
||||
async with (
|
||||
tractor.open_root_actor(
|
||||
|
|
@ -864,15 +619,5 @@ def test_moc_reentry_during_teardown(
|
|||
):
|
||||
tn.start_soon(use_and_exit)
|
||||
tn.start_soon(reenter_during_teardown)
|
||||
await reentry_started.wait()
|
||||
|
||||
# Wait until the re-entry task is queued on MOC's
|
||||
# per-key lock while `_Cache.run_ctx()` remains
|
||||
# blocked in the first generation's `__aexit__()`.
|
||||
# Only then release cleanup, making the intended
|
||||
# enter-during-sibling-exit ordering deterministic.
|
||||
await wait_all_tasks_blocked()
|
||||
assert not first_done.is_set()
|
||||
allow_aexit.set()
|
||||
|
||||
trio.run(main)
|
||||
|
|
|
|||
|
|
@ -198,6 +198,9 @@ class Channel:
|
|||
# assert transport.raddr == addr
|
||||
chan = Channel(transport=transport)
|
||||
|
||||
# ?TODO, compact this into adapter level-methods?
|
||||
# -[ ] would avoid extra repr-calcs if level not active?
|
||||
# |_ how would the `calc_if_level` look though? func?
|
||||
if log.at_least_level('runtime'):
|
||||
from tractor.devx import (
|
||||
pformat as _pformat,
|
||||
|
|
@ -322,8 +325,6 @@ class Channel:
|
|||
'''
|
||||
__tracebackhide__: bool = hide_tb
|
||||
try:
|
||||
if log.at_least_level('transport'):
|
||||
# don't materialize the payload repr if not necessary
|
||||
log.transport(
|
||||
'=> send IPC msg:\n\n'
|
||||
f'{pformat(payload)}\n'
|
||||
|
|
|
|||
|
|
@ -309,10 +309,7 @@ class MsgpackTransport(MsgTransport):
|
|||
log.transport(f'received header {size}') # type: ignore
|
||||
msg_bytes: bytes = await self.recv_stream.receive_exactly(size)
|
||||
|
||||
if log.at_least_level('transport'):
|
||||
log.transport( # type: ignore
|
||||
f'received {msg_bytes}'
|
||||
)
|
||||
log.transport(f"received {msg_bytes}") # type: ignore
|
||||
try:
|
||||
# NOTE: lookup the `trio.Task.context`'s var for
|
||||
# the current `MsgCodec`.
|
||||
|
|
|
|||
|
|
@ -111,7 +111,9 @@ def at_least_level(
|
|||
if isinstance(level, str):
|
||||
level: int = CUSTOM_LEVELS[level.upper()]
|
||||
|
||||
return log.isEnabledFor(level)
|
||||
if log.getEffectiveLevel() <= level:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# TODO, compare with using a "filter" instead?
|
||||
|
|
|
|||
|
|
@ -306,12 +306,12 @@ class PldRx(Struct):
|
|||
):
|
||||
try:
|
||||
pld: PayloadT = self._pld_dec.decode(pld)
|
||||
if log.at_least_level('runtime'):
|
||||
# don't materialize the payload repr if not necessary
|
||||
log.runtime(
|
||||
f'Decoded payload for\n'
|
||||
f'\n'
|
||||
# f'\n'
|
||||
f'{msg}\n'
|
||||
# ^TODO?, ideally just render with `,
|
||||
# pld={decode}` in the `msg.pformat()`??
|
||||
f'where, '
|
||||
f'{type(msg).__name__}.pld={pld!r}\n'
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1003,15 +1003,17 @@ async def process_messages(
|
|||
task_status.started(loop_cs)
|
||||
|
||||
async for msg in chan:
|
||||
if log.at_least_level('transport'):
|
||||
log.transport( # type: ignore
|
||||
f'IPC msg from peer\n'
|
||||
f'<= {chan.aid.reprol()}\n\n'
|
||||
|
||||
# TODO: pretty-printing structs is FRAGILE;
|
||||
# -[ ] add a non-raising log formatter with
|
||||
# native-repr fallback before using
|
||||
# `.msg.pretty_struct` here.
|
||||
# TODO: use of the pprinting of structs is
|
||||
# FRAGILE and should prolly not be
|
||||
#
|
||||
# avoid fmting depending on loglevel for perf?
|
||||
# -[ ] specifically `pretty_struct.pformat()` sub-call..?
|
||||
# - how to only log-level-aware actually call this?
|
||||
# -[ ] use `.msg.pretty_struct` here now instead!
|
||||
# f'{pretty_struct.pformat(msg)}\n'
|
||||
f'{msg}\n'
|
||||
)
|
||||
|
|
@ -1260,7 +1262,6 @@ async def process_messages(
|
|||
log.exception(message)
|
||||
raise RuntimeError(message)
|
||||
|
||||
if log.at_least_level('transport'):
|
||||
log.transport(
|
||||
'Waiting on next IPC msg from\n'
|
||||
f'peer: {chan.aid.reprol()}\n'
|
||||
|
|
|
|||
|
|
@ -198,16 +198,6 @@ async def gather_contexts(
|
|||
# Further potential examples of interest:
|
||||
# https://gist.github.com/njsmith/cf6fc0a97f53865f2c671659c88c1798#file-cache-py-L8
|
||||
|
||||
class _CtxExit:
|
||||
'''
|
||||
Completion state for a cached context's shared exit.
|
||||
|
||||
'''
|
||||
def __init__(self) -> None:
|
||||
self.done = trio.Event()
|
||||
self.error: Exception|None = None
|
||||
|
||||
|
||||
class _Cache:
|
||||
'''
|
||||
Globally (actor-processs scoped) cached, task access to
|
||||
|
|
@ -223,11 +213,7 @@ class _Cache:
|
|||
values: dict[Any, Any] = {}
|
||||
resources: dict[
|
||||
Hashable,
|
||||
tuple[
|
||||
trio.Nursery,
|
||||
trio.Event,
|
||||
_CtxExit,
|
||||
],
|
||||
tuple[trio.Nursery, trio.Event]
|
||||
] = {}
|
||||
# nurseries: dict[int, trio.Nursery] = {}
|
||||
no_more_users: trio.Event|None = None
|
||||
|
|
@ -237,39 +223,19 @@ class _Cache:
|
|||
cls,
|
||||
mng,
|
||||
ctx_key: tuple,
|
||||
ctx_exit: _CtxExit,
|
||||
task_status: trio.TaskStatus[T] = trio.TASK_STATUS_IGNORED,
|
||||
|
||||
) -> None:
|
||||
entered: bool = False
|
||||
try:
|
||||
async with mng as value:
|
||||
entered = True
|
||||
(
|
||||
_,
|
||||
no_more_users,
|
||||
_,
|
||||
) = cls.resources[ctx_key]
|
||||
_, no_more_users = cls.resources[ctx_key]
|
||||
cls.values[ctx_key] = value
|
||||
task_status.started(value)
|
||||
try:
|
||||
await no_more_users.wait()
|
||||
finally:
|
||||
cls.values.pop(ctx_key)
|
||||
value = cls.values.pop(ctx_key)
|
||||
cls.resources.pop(ctx_key)
|
||||
|
||||
except Exception as exc:
|
||||
if not entered:
|
||||
raise
|
||||
|
||||
# Deliver regular `__aexit__()` failures to the final
|
||||
# consumer instead of raising into the service nursery.
|
||||
ctx_exit.error = exc
|
||||
|
||||
finally:
|
||||
if entered:
|
||||
ctx_exit.done.set()
|
||||
|
||||
|
||||
class _UnresolvedCtx:
|
||||
'''
|
||||
|
|
@ -315,10 +281,9 @@ async def maybe_open_context(
|
|||
)
|
||||
|
||||
# yielded output
|
||||
# sentinel = object()
|
||||
yielded: Any = _UnresolvedCtx
|
||||
user_registered: bool = False
|
||||
ctx_exit: _CtxExit|None = None
|
||||
exit_error: Exception|None = None
|
||||
|
||||
# Lock resource acquisition around task racing / ``trio``'s
|
||||
# scheduler protocol.
|
||||
|
|
@ -335,6 +300,7 @@ async def maybe_open_context(
|
|||
] = trio.StrictFIFOLock()
|
||||
header: str = 'Allocated NEW lock for @acm_func,\n'
|
||||
else:
|
||||
await trio.lowlevel.checkpoint()
|
||||
header: str = 'Reusing OLD lock for @acm_func,\n'
|
||||
|
||||
log.debug(
|
||||
|
|
@ -402,6 +368,7 @@ async def maybe_open_context(
|
|||
resources = _Cache.resources
|
||||
entry: tuple|None = resources.get(ctx_key)
|
||||
if entry:
|
||||
service_tn, ev = entry
|
||||
raise RuntimeError(
|
||||
f'Caching resources ALREADY exist?!\n'
|
||||
f'ctx_key={ctx_key!r}\n'
|
||||
|
|
@ -409,18 +376,12 @@ async def maybe_open_context(
|
|||
f'task: {task}\n'
|
||||
)
|
||||
|
||||
ctx_exit = _CtxExit()
|
||||
resources[ctx_key] = (
|
||||
service_tn,
|
||||
trio.Event(),
|
||||
ctx_exit,
|
||||
)
|
||||
resources[ctx_key] = (service_tn, trio.Event())
|
||||
try:
|
||||
yielded: Any = await service_tn.start(
|
||||
_Cache.run_ctx,
|
||||
mngr,
|
||||
ctx_key,
|
||||
ctx_exit,
|
||||
)
|
||||
except BaseException:
|
||||
# If `run_ctx` (wrapping the acm's `__aenter__`)
|
||||
|
|
@ -466,11 +427,6 @@ async def maybe_open_context(
|
|||
raise taskc
|
||||
else:
|
||||
# XXX, cached-entry-path
|
||||
(
|
||||
_,
|
||||
_,
|
||||
ctx_exit,
|
||||
) = _Cache.resources[ctx_key]
|
||||
_Cache.users[ctx_key] += 1
|
||||
user_registered = True
|
||||
log.debug(
|
||||
|
|
@ -489,17 +445,24 @@ async def maybe_open_context(
|
|||
)
|
||||
|
||||
finally:
|
||||
if lock.locked():
|
||||
stats: trio.LockStatistics = lock.statistics()
|
||||
owner: trio.Task|None = stats.owner
|
||||
log.error(
|
||||
f'Lock never released by last owner={owner!r} !?\n'
|
||||
f'{stats}\n'
|
||||
f'\n'
|
||||
f'task={task!r}\n'
|
||||
f'ctx_key={ctx_key!r}\n'
|
||||
f'acm_func={acm_func}\n'
|
||||
|
||||
)
|
||||
|
||||
if user_registered:
|
||||
# Serialize user registration and teardown under the same
|
||||
# per-key lock so no entrant can acquire a resource after
|
||||
# its final user has committed to exiting it.
|
||||
with trio.CancelScope(shield=True):
|
||||
await lock.acquire()
|
||||
try:
|
||||
_Cache.users[ctx_key] -= 1
|
||||
|
||||
# If no consumers remain, keep entrants queued
|
||||
# until the cached context has completely exited.
|
||||
if yielded is not _UnresolvedCtx:
|
||||
# if no more consumers, teardown the client
|
||||
if _Cache.users[ctx_key] <= 0:
|
||||
log.debug(
|
||||
f'De-allocating @acm-func entry\n'
|
||||
|
|
@ -507,39 +470,19 @@ async def maybe_open_context(
|
|||
f'acm_func={acm_func!r}\n'
|
||||
)
|
||||
|
||||
# XXX: if we're cancelled, the entry may
|
||||
# have never been entered since the nursery
|
||||
# task was killed.
|
||||
# XXX: if we're cancelled we the entry may have never
|
||||
# been entered since the nursery task was killed.
|
||||
# _, no_more_users = _Cache.resources[ctx_key]
|
||||
entry = _Cache.resources.get(ctx_key)
|
||||
if entry:
|
||||
(
|
||||
_,
|
||||
no_more_users,
|
||||
ctx_exit,
|
||||
) = entry
|
||||
_, no_more_users = entry
|
||||
no_more_users.set()
|
||||
|
||||
assert ctx_exit is not None
|
||||
await ctx_exit.done.wait()
|
||||
exit_error = ctx_exit.error
|
||||
|
||||
# A queued entrant already holds a reference
|
||||
# to this lock. Keep it registered until that
|
||||
# task has acquired and released it.
|
||||
stats = lock.statistics()
|
||||
if not stats.tasks_waiting:
|
||||
maybe_lock = _Cache.locks.get(ctx_key)
|
||||
if maybe_lock is lock:
|
||||
_Cache.locks.pop(ctx_key)
|
||||
else:
|
||||
log.error(
|
||||
f'Resource lock for {ctx_key} '
|
||||
f'was replaced before teardown?'
|
||||
maybe_lock = _Cache.locks.pop(
|
||||
ctx_key,
|
||||
None,
|
||||
)
|
||||
if maybe_lock is None:
|
||||
log.error(
|
||||
f'Resource lock for {ctx_key} ALREADY POPPED?'
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
if exit_error is not None:
|
||||
# Always re-raise a regular `__aexit__()` error at the
|
||||
# final consumer's context boundary.
|
||||
raise exit_error
|
||||
|
|
|
|||
Loading…
Reference in New Issue