Add new timeout for observe_value

[coretl]

@DominicOram @olliesilvester this is the epics test I mentioned in DiamondLightSource/dodal#897 (comment)

It looks like I need to yield after the q.get rather than before to let the timeout break without an extra yield

[coretl]

Spoke too soon, doesn't work on CI...

[coretl]

Ok, looks like there are actually numerous python bugs in asyncio.wait_for, which is fixed by its reimplementation on top of asyncio.timeout in 3.12: python/cpython#96764

I think this fixes your particular use case, but you can't pass a timeout to observe_value and also wrap it in wait_for with a timeout on python 3.11 and below.

I would also suggest that you move the opencv calls to another thread and consume all the values from the Queue between the two, dropping old updates.

[olliesilvester]

Great, thank you!

[coretl]

Ok, still some issues will this, will investigate next week

[coretl]

@DominicOram @olliesilvester @rtuck99 I think this is now working, but would be good to talk through the code with one of you to check I understand the problem correctly

[olliesilvester]

@DominicOram @olliesilvester @rtuck99 I think this is now working, but would be good to talk through the code with one of you to check I understand the problem correctly

I'm about today, I'm just reading through the changes now

[olliesilvester]

Discussed with @coretl, I think this change fixes our issue but I will investigate further to see if there are any safer solutions

[DominicOram]

This looks reasonable to me, especially if @olliesilvester has talked over it too. My only comment would be that if we think it's mostly a problem with asyncio.wait_for in <3.11 then we should add a comment/issue to drop it when we drop 3.11. But if we think there are other, more general, problems this might catch then maybe we shouldn't drop it.

[olliesilvester]

I tried using asyncio.timeout instead, and we seem to get the same issue. I guess this explains why python 3.12 still had this issue.

I'm leaning towards just not using asyncio.wait_for for this scenario. We could just do

async def observe_value(...):
  if time.time() - start_time > overall_timeout:
                  raise TimeoutError
...

instead, and warn against using asyncio.wait_for (or timeout) with observe_value

[coretl]

I'm leaning towards just not using asyncio.wait_for for this scenario

It's messy, but it's probably the most pragmatic solution here, then we relegate wait_for_pending_wakeups to ophyd_async.testing and only use it in tests

@DominicOram thoughts?

@olliesilvester when this is decided would you mind amending this PR or making a new one with the implementation please?

[DominicOram]

I'm not a big fan of this, having to roll you're own timeout code when there's provision for it in asyncio looks on the face of it like a code-smell if you're not familiar with the bug. How would we feel about adding the functionality into observe_value itself e.g. optionally pass in a overall_timeout value? This would then make the path of least resistance be to just use that

[olliesilvester]

Yeah, the timeout will be in observe_value as an optional parameter

[olliesilvester]

I've given this a go but there are some extra things to think about

1. The regular timeout defaults to None, which means that if we do our overall_timeout, as above, then the code can get stuck forever doing get_value. asyncio.wait_for protects against this, regardless of the other problem it caused

Solution here is either to either give timeout a default value or make a clever decorator which automatically supplies our overall_timeout parameter with the same value which was used in our timeout in asyncio.wait_for(timeout=...)

2. If we don't go with the latter solution above, then we could add a warning (with a decorator?) if observe_value has been used with asyncio.timeout

I think the clever decorator is best here, assuming it isn't too hard to make

Thoughts @coretl?

[coretl]

I suggest something like:

async def observe_value(signal, timeout=None, overall_timeout=None):
    overall_deadline = time.monotonic() + overall_timeout if overall_timeout else None
    ...
    while True:
        deadline = time.monotonic() + timeout if timeout else overall_deadline
        if deadline and overall_deadline:
            deadline = min(deadline, overall_deadline)
        value = await asyncio.wait_for(q.get(), deadline - time.monotonic() if deadline else 0) 
    ...

[olliesilvester]

Although people will probably still do asyncio.wait_for(observe_value...), this issue (if anyone even sees it again) should be easier to spot with the extra docstring