Master branch bug with `held_for` parameter

[mozesa]

Hello John,

I would like to ask for your help.

I am facing with the following problem which I think comes from the negation of the given predicate. So from this line: result = self._level_results[lambda v: not predicate(v)] The result is that even though the predicate has been evaluated True, it keeps checking the negation of the predicate.

This is my clue.

    @value.setter
    def value(self, x):
        print(f"{self._level_results=}")  # NEW
        if self._value != x:
            old = self._old = self._value
            new = self._value = x
            for f, result in list(self._level_results.items()):
                if f(new):
                    result.value = new
                    result.event.unpark_all()
                    del self._level_results[f]
            for f, result in list(self._edge_results.items()):
                if f(new, old):
                    result.value = (new, old)
                    result.event.unpark_all()
                    del self._edge_results[f]

    async def wait_value(self, predicate, *, held_for=0):
        while True:
            if not predicate(self._value):
                result = self._level_results[predicate]
                await result.event.park()
                value = result.value
            else:
                value = self._value
                await trio.sleep(0)
            if held_for > 0:
                with trio.move_on_after(held_for):
                    if predicate(self._value):
                        result = self._level_results[lambda v: not predicate(v)]
                        await result.event.park()
                    print("before continue")  # NEW
                    continue
            print("before break")  # NEW
            break
        return value

held_for=1.0
self._level_results=defaultdict(<class 'trio_util._async_value._Result'>, {<bound method VWNeoNutThreadChecking.check_bit_head of ...>})
self._level_results=defaultdict(<class 'trio_util._async_value._Result'>, {<function AsyncValue.wait_value.<locals>.<lambda> at 0x00000129738DD040>: <trio_util._async_value._Result object at 0x000001297392C880>})
before break
self._level_results=defaultdict(<class 'trio_util._async_value._Result'>, {<function AsyncValue.wait_value.<locals>.<lambda> at 0x00000129738DD040>: <trio_util._async_value._Result object at 0x000001297392C880>})

held_for=0.0
self._level_results=defaultdict(<class 'trio_util._async_value._Result'>, {<bound method VWNeoNutThreadChecking.check_bit_head of ...>})
before break
self._level_results=defaultdict(<class 'trio_util._async_value._Result'>, {})

[belm0]

Would you clarify what the bug is? It sounds like wait_value(..., held_for=...) is working as intended.

Why would it be a problem for your predicate to be evaluated after the wait_value() returns? It's an implementation detail. Predicate functions are typically a pure expression of the input args, and should not have side effects.

[mozesa]

I found it strange as I didn't expect to have a function getting called later by a function having passed.

I mean trio_util dels the predicate so it should just do the same for the negation of the predicate as well.

I use a device which has to be instantiated at modul level and only once during the lifetime of the app (something like a singleton). My main concern is that the more negations of predicate I have, the more time it needs for the evaluation b/c of O(N) complexity.

I really appreciate your help and the effort you putting into this package.

Great thanks!

[belm0]

I see, thank you for clarifying.

I can delete the held_for predicate, but it would affect efficiency a little because there can be a race condition with the other task. (It must be a "safe" delete with testing that the key exists first.)

But there are other situations where this could happen, so 1) if a guarantee is made about scope of predicate eval then it should be made across the API; and 2) I'm not sure yet if this is a good value (important to users and worth complexity and runtime cost).

Eval of predicate outside the scope of the method call could happen in any cancel scope context (just like the case of the wait_value implementation):

x = AsyncValue(0)

with trio.move_on_after(1):
    await x.wait_value(f)

x.value = 10

In this example, the await x.wait_value() coroutine will be cancelled, and the predicate remains. f() will be called when the value 10 is assigned. So this matter has a larger scope than held_for.

[belm0]

To address this generally, the values in the predicate maps would need to be ref counted. Rather than the value setter deleting entries, they would be deleted by the wait_*() method when the last ref was removed.

For wait_value():

if not predicate(self._value):
    result = self._level_results[predicate]
    result.refs += 1
    try:
        await result.event.park()
    finally:
        result.refs -= 1
        if not result.refs:
            del self._level_results[predicate]
    value = result.value

[mozesa]

Now I see! Thanks for detailed explanation.

You are absolutely right cancelled coroutines leaving the predicate intact. So yes it is something which should definitly be addressed generally.

The solution you draw up is clever and works well (I have checked it).

This package has already been a must-have. 👍

[belm0]

The fix had an unintended consequence: it changed some undocumented semantics for the case where value has multiple assignments matching the predicate before the waiting task is resumed.

Previously the initial match was always returned, because the value setter atomically removed the predicate when unparking waiting tasks. Now it is undefined as to which value will be returned.

Either semantics seem valid, but it was nice to have deterministic behavior...