Compare commits

..

5 Commits

Author SHA1 Message Date
Gud Boi 06c4af17e4 Fix `BroadcastReceiver` lag counts
`BroadcastReceiver.receive_nowait()` treated `seq` as a deque index
but subtracted `BroadcastState.maxlen` without counting the first
invalid index. A one-slot queue thus claimed it dropped zero values
after its subscriber missed one.

Include that first displaced value in the count. Preserve the
existing Tokio-style reset to the oldest retained item.

Also, cover exact loss reporting and recovery for one- and
three-slot retention windows.

Prompt-IO: ai/prompt-io/opencode/20260811T233833Z_7cbd64ee_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
2026-08-11 19:45:53 -04:00
Bd 83b3488455
Merge pull request #488 from goodboy/wkt/moc_teardown_completion
Wait for ctx exit in `maybe_open_context()`
2026-08-11 12:16:15 -04:00
Gud Boi daa661aba3 Clarify `maybe_open_context()` teardown notes
Drop the stale sentinel experiment and fix the cancellation-path
comment. Document that cached regular `__aexit__()` failures are
always re-raised at the final consumer boundary.

Review: PR #488 (goodboy)
https://github.com/goodboy/tractor/pull/488

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-11 11:38:43 -04:00
Gud Boi 55ec3dbf51 Drop unused `_Cache` teardown bindings
Remove the unused `value` assignment after cache eviction and skip
unpacking stale resource state before raising its invariant error.

Review: PR #488 (Copilot)
https://github.com/goodboy/tractor/pull/488#pullrequestreview-4850557500

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-07 22:24:27 -04:00
Gud Boi a753625fc6 Wait for ctx exit in `maybe_open_context()`
Block the final user on its cached resource's `__aexit__()` and
raise regular cleanup errors at that user's ctx boundary.

Deats,
- serialize user registration and teardown under each cache-key lock
- keep queued entrants on the same lock through resource replacement
- preserve `Cancelled`, `KeyboardInterrupt`, and `SystemExit` flow
- cover successful, failing, cancelled, and re-entry teardown paths

Prompt-IO: ai/prompt-io/opencode/20260804T030309Z_65bf9df5_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
2026-08-03 23:45:42 -04:00
9 changed files with 612 additions and 78 deletions

View File

@ -0,0 +1,39 @@
---
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.

View File

@ -0,0 +1,36 @@
---
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`.

View File

@ -0,0 +1,30 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: ses_0799212ebffe42arY96czXn89F
timestamp: 2026-08-11T23:38:33Z
git_ref: 7cbd64ee
scope: code
substantive: true
raw_file: 20260811T233833Z_7cbd64ee_prompt_io.raw.md
---
## Prompt
Open a new isolated worktree in the local Tractor repository and draft a fix
for `BroadcastReceiver` reporting that a lagged one-slot consumer dropped
zero values when one value had actually been displaced.
## Response summary
Corrected the off-by-one lag count while preserving cursor recovery and added
deterministic narrow- and wider-window regressions for exact loss reporting.
## Files changed
- `tractor/trionics/_broadcast.py` - exact broadcast overrun count.
- `tests/test_task_broadcasting.py` - lag count and recovery regression.
## Human edits
None - generated output follows the user's diagnosed edge case.

View File

@ -0,0 +1,50 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-11T23:38:33Z
git_ref: 7cbd64ee
diff_cmd: git diff HEAD~1..HEAD
---
The user asked for a Tractor fix in a new isolated worktree after a live piker
failure reported:
```text
tractor.trionics._broadcast.Lagged:
Task `piker.brokers.ib.broker.handle_order_requests` overrun and
dropped `0` values
```
Inspection showed the lag exception was valid but its count was off by one.
`BroadcastReceiver.receive_nowait()` treats `seq` as a deque index. With a
one-entry queue, index zero is the only retained value and `seq == 1` already
means one value was displaced. The old `seq - maxlen` calculation therefore
reported zero instead of one.
> `git diff HEAD~1..HEAD -- tractor/trionics/_broadcast.py`
Adjusted the lag count to `seq - maxlen + 1` and documented why the first
invalid deque index must be included. The existing Tokio-style cursor reset
remains unchanged.
> `git diff HEAD~1..HEAD -- tests/test_task_broadcasting.py`
Added a deterministic parameterized regression covering a one-slot queue
with one dropped value and a three-slot queue with two dropped values. The
test keeps the root receiver idle while a child subscriber synchronously
drains each produced value, asserts the exact `Lagged` message, and proves the
next receive resumes at the oldest retained item.
Verification output:
```text
.. [100%]
2 passed in 0.04s
.......... [100%]
10 passed in 4.65s
```
The targeted import resolved to the new Tractor worktree. Python compilation
and `git diff --check` passed. Adversarial review found no actionable issues;
zero-capacity channels remain a pre-existing untested edge outside this fix.

View File

@ -0,0 +1,22 @@
# 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.

View File

@ -9,6 +9,7 @@ from typing import Awaitable
import pytest import pytest
import trio import trio
from trio.testing import wait_all_tasks_blocked
import tractor import tractor
from tractor.trionics import ( from tractor.trionics import (
maybe_open_context, maybe_open_context,
@ -94,6 +95,232 @@ def test_resource_only_entered_once(key_on):
trio.run(main) 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 @tractor.context
async def streamer( async def streamer(
ctx: tractor.Context, ctx: tractor.Context,
@ -548,43 +775,52 @@ def test_moc_reentry_during_teardown(
loglevel: str, loglevel: str,
): ):
''' '''
Reproduce the piker `open_cached_client('kraken')` race: Reproduce re-entry while an identical cached context exits.
- same `acm_func`, NO kwargs (identical `ctx_key`) - multiple tasks use the same `acm_func` with no kwargs,
- multiple tasks share the cached resource producing an identical `ctx_key`;
- all users exit -> teardown starts - all users leave and the final user starts resource teardown;
- a NEW task enters during `_Cache.run_ctx.__aexit__` - `_Cache.run_ctx()` removes the cached value and resource entry
- `values[ctx_key]` is gone (popped in inner finally) before entering the resource's blocking `__aexit__()` body;
but `resources[ctx_key]` still exists (outer finally - a new task attempts to enter that same `ctx_key` during exit;
hasn't run yet bc the acm cleanup has checkpoints) - the per-key lock keeps that entrant queued until exit
- old code: `assert not resources.get(ctx_key)` FIRES completes;
- the entrant then receives a fresh cache miss and resource.
This models the real-world scenario where `brokerd.kraken` Without teardown sharing the registration lock, re-entry could
tasks concurrently call `open_cached_client('kraken')` race resource replacement while the prior generation was still
(same `acm_func`, empty kwargs, shared `ctx_key`) and exiting. The final user could also return before that exit
the teardown/re-entry race triggers intermittently. 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.
''' '''
async def main(): async def main():
in_aexit = trio.Event() in_aexit = trio.Event()
allow_aexit = trio.Event()
reentry_started = trio.Event()
generation: int = 0
@acm @acm
async def cached_client(): async def cached_client():
''' '''
Simulates `kraken.api.get_client()`: Simulate a no-argument `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' yield 'the-client'
# Signal that we're in __aexit__ — at this if resource_generation == 1:
# point `values` has already been popped by in_aexit.set()
# `run_ctx`'s inner finally, but `resources` await allow_aexit.wait()
# is still alive (outer finally hasn't run).
in_aexit.set()
await trio.sleep(10)
first_done = trio.Event() first_done = trio.Event()
@ -598,16 +834,25 @@ def test_moc_reentry_during_teardown(
async def reenter_during_teardown(): async def reenter_during_teardown():
''' '''
Wait for the acm's `__aexit__` to start (meaning Wait for the acm's `__aexit__` to start (meaning
`values` is popped but `resources` still exists), the cached value is no longer available), then re-enter.
then re-enter triggering the assert.
''' '''
await in_aexit.wait() 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( async with maybe_open_context(
cached_client, cached_client,
) as (cache_hit, value): ) as (cache_hit, value):
assert not cache_hit
assert value == 'the-client' assert value == 'the-client'
await first_done.wait()
with trio.fail_after(5): with trio.fail_after(5):
async with ( async with (
tractor.open_root_actor( tractor.open_root_actor(
@ -619,5 +864,15 @@ def test_moc_reentry_during_teardown(
): ):
tn.start_soon(use_and_exit) tn.start_soon(use_and_exit)
tn.start_soon(reenter_during_teardown) 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) trio.run(main)

View File

@ -307,6 +307,48 @@ def test_subscribe_errors_after_close():
trio.run(main) trio.run(main)
@pytest.mark.parametrize(
('size', 'sent', 'dropped'),
[
(1, 2, 1),
(3, 5, 2),
],
)
def test_lagged_reports_exact_drop_count(
size: int,
sent: int,
dropped: int,
) -> None:
'''
`Lagged` must report every value outside the retained window.
`BroadcastReceiver.receive_nowait()` previously subtracted the
queue length from an already-invalid deque index without counting
that first displaced value. A one-slot queue therefore claimed it
dropped zero values after two sends. Keep one root receiver idle
while a child subscriber drains every produced value, then prove
the lag error reports the exact overrun and positions the root at
the oldest value still retained by `BroadcastState.queue`.
'''
async def main() -> None:
tx, rx = trio.open_memory_channel(size)
brx = broadcast_receiver(rx, size)
async with brx.subscribe() as fast:
for value in range(sent):
await tx.send(value)
assert await fast.receive() == value
match = rf'dropped `{dropped}` values'
with pytest.raises(Lagged, match=match):
await brx.receive()
assert await brx.receive() == sent - size
trio.run(main)
def test_ensure_slow_consumers_lag_out( def test_ensure_slow_consumers_lag_out(
reg_addr, reg_addr,
start_method, start_method,

View File

@ -237,7 +237,10 @@ class BroadcastReceiver(ReceiveChannel):
# https://docs.rs/tokio/1.11.0/tokio/sync/broadcast/index.html#lagging # https://docs.rs/tokio/1.11.0/tokio/sync/broadcast/index.html#lagging
mxln = state.maxlen mxln = state.maxlen
lost = seq - mxln # `seq == mxln` is already one past the final
# valid deque index, so include that first
# displaced value in the loss count.
lost = seq - mxln + 1
# decrement to the last value and expect # decrement to the last value and expect
# consumer to either handle the ``Lagged`` and come back # consumer to either handle the ``Lagged`` and come back

View File

@ -198,6 +198,16 @@ async def gather_contexts(
# Further potential examples of interest: # Further potential examples of interest:
# https://gist.github.com/njsmith/cf6fc0a97f53865f2c671659c88c1798#file-cache-py-L8 # 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: class _Cache:
''' '''
Globally (actor-processs scoped) cached, task access to Globally (actor-processs scoped) cached, task access to
@ -213,7 +223,11 @@ class _Cache:
values: dict[Any, Any] = {} values: dict[Any, Any] = {}
resources: dict[ resources: dict[
Hashable, Hashable,
tuple[trio.Nursery, trio.Event] tuple[
trio.Nursery,
trio.Event,
_CtxExit,
],
] = {} ] = {}
# nurseries: dict[int, trio.Nursery] = {} # nurseries: dict[int, trio.Nursery] = {}
no_more_users: trio.Event|None = None no_more_users: trio.Event|None = None
@ -223,18 +237,38 @@ class _Cache:
cls, cls,
mng, mng,
ctx_key: tuple, ctx_key: tuple,
ctx_exit: _CtxExit,
task_status: trio.TaskStatus[T] = trio.TASK_STATUS_IGNORED, task_status: trio.TaskStatus[T] = trio.TASK_STATUS_IGNORED,
) -> None: ) -> None:
async with mng as value: entered: bool = False
_, no_more_users = cls.resources[ctx_key] try:
cls.values[ctx_key] = value async with mng as value:
task_status.started(value) entered = True
try: (
await no_more_users.wait() _,
finally: no_more_users,
value = cls.values.pop(ctx_key) _,
cls.resources.pop(ctx_key) ) = 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)
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: class _UnresolvedCtx:
@ -281,9 +315,10 @@ async def maybe_open_context(
) )
# yielded output # yielded output
# sentinel = object()
yielded: Any = _UnresolvedCtx yielded: Any = _UnresolvedCtx
user_registered: bool = False user_registered: bool = False
ctx_exit: _CtxExit|None = None
exit_error: Exception|None = None
# Lock resource acquisition around task racing / ``trio``'s # Lock resource acquisition around task racing / ``trio``'s
# scheduler protocol. # scheduler protocol.
@ -300,7 +335,6 @@ async def maybe_open_context(
] = trio.StrictFIFOLock() ] = trio.StrictFIFOLock()
header: str = 'Allocated NEW lock for @acm_func,\n' header: str = 'Allocated NEW lock for @acm_func,\n'
else: else:
await trio.lowlevel.checkpoint()
header: str = 'Reusing OLD lock for @acm_func,\n' header: str = 'Reusing OLD lock for @acm_func,\n'
log.debug( log.debug(
@ -368,7 +402,6 @@ async def maybe_open_context(
resources = _Cache.resources resources = _Cache.resources
entry: tuple|None = resources.get(ctx_key) entry: tuple|None = resources.get(ctx_key)
if entry: if entry:
service_tn, ev = entry
raise RuntimeError( raise RuntimeError(
f'Caching resources ALREADY exist?!\n' f'Caching resources ALREADY exist?!\n'
f'ctx_key={ctx_key!r}\n' f'ctx_key={ctx_key!r}\n'
@ -376,12 +409,18 @@ async def maybe_open_context(
f'task: {task}\n' f'task: {task}\n'
) )
resources[ctx_key] = (service_tn, trio.Event()) ctx_exit = _CtxExit()
resources[ctx_key] = (
service_tn,
trio.Event(),
ctx_exit,
)
try: try:
yielded: Any = await service_tn.start( yielded: Any = await service_tn.start(
_Cache.run_ctx, _Cache.run_ctx,
mngr, mngr,
ctx_key, ctx_key,
ctx_exit,
) )
except BaseException: except BaseException:
# If `run_ctx` (wrapping the acm's `__aenter__`) # If `run_ctx` (wrapping the acm's `__aenter__`)
@ -427,6 +466,11 @@ async def maybe_open_context(
raise taskc raise taskc
else: else:
# XXX, cached-entry-path # XXX, cached-entry-path
(
_,
_,
ctx_exit,
) = _Cache.resources[ctx_key]
_Cache.users[ctx_key] += 1 _Cache.users[ctx_key] += 1
user_registered = True user_registered = True
log.debug( log.debug(
@ -445,44 +489,57 @@ async def maybe_open_context(
) )
finally: 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: if user_registered:
_Cache.users[ctx_key] -= 1 # 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 yielded is not _UnresolvedCtx: # If no consumers remain, keep entrants queued
# if no more consumers, teardown the client # until the cached context has completely exited.
if _Cache.users[ctx_key] <= 0: if _Cache.users[ctx_key] <= 0:
log.debug( log.debug(
f'De-allocating @acm-func entry\n' f'De-allocating @acm-func entry\n'
f'ctx_key={ctx_key!r}\n' f'ctx_key={ctx_key!r}\n'
f'acm_func={acm_func!r}\n' f'acm_func={acm_func!r}\n'
) )
# XXX: if we're cancelled we the entry may have never # XXX: if we're cancelled, the entry may
# been entered since the nursery task was killed. # have never been entered since the nursery
# _, no_more_users = _Cache.resources[ctx_key] # task was killed.
entry = _Cache.resources.get(ctx_key) entry = _Cache.resources.get(ctx_key)
if entry: if entry:
_, no_more_users = entry (
no_more_users.set() _,
no_more_users,
ctx_exit,
) = entry
no_more_users.set()
maybe_lock = _Cache.locks.pop( assert ctx_exit is not None
ctx_key, await ctx_exit.done.wait()
None, exit_error = ctx_exit.error
)
if maybe_lock is None: # A queued entrant already holds a reference
log.error( # to this lock. Keep it registered until that
f'Resource lock for {ctx_key} ALREADY POPPED?' # 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?'
)
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