python - 等待条件变量超时 : lock not reacquired in time

Question

我有一个名为 cond 的 asyncio.Condition。我想等，但只能等那么久，然后就放弃了。由于asyncio.Condition.wait不会超时，因此不能直接完成此操作。 The docs声明应该使用 asyncio.wait_for 来包装并提供超时:

The asyncio.wait_for() function can be used to cancel a task after a timeout.

因此，我们得出以下解决方案:

async def coro():
    print("Taking lock...")
    async with cond:
        print("Lock acquired.")
        print("Waiting!")
        await asyncio.wait_for(cond.wait(), timeout=999)
        print("Was notified!")
    print("Lock released.")

<小时/>

现在假设coro本身在运行五秒后被取消。这会在 wait_for 中引发 CancelledError，从而在重新引发错误之前取消 cond.wait。然后，错误会传播到 coro，由于 async with block ，它会隐式尝试释放 cond 中的锁。然而，当前没有持有锁； cond.wait 已被取消，但没有机会处理该取消并重新获取锁。因此，我们得到一个丑陋的异常，如下所示:

Taking lock...
Lock acquired.
Waiting!
ERROR:asyncio:Task exception was never retrieved
future: <Task finished coro=<coro() done, defined at [REDACTED]> exception=RuntimeError('Lock is not acquired.',)>
Traceback (most recent call last):
  [REDACTED], in coro
    await asyncio.wait_for(cond.wait(), timeout=999)
  [REDACTED], in wait_for
    yield from waiter
concurrent.futures._base.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  [REDACTED], in coro
    print("Was notified!")
  [REDACTED], in coro
    res = func(*args, **kw)
  [REDACTED], in __aexit__
    self.release()
  [REDACTED], in release
    raise RuntimeError('Lock is not acquired.')
RuntimeError: Lock is not acquired.

换句话说，在处理 CancelledError 时，coro 由于尝试释放未持有的锁而引发了 RuntimeError。堆栈跟踪显示 print("Was notification!") 行的原因是因为这是有问题的 async with block 的最后一行。

<小时/><删除>这感觉不像是我能解决的问题；我开始怀疑这是库本身的错误。但是，我想不出任何方法来避免该问题或创建解决方法，因此任何想法将不胜感激。

在编写这个问题并进一步调查时，我在 Python 错误跟踪器上偶然发现了类似的问题，最终检查了 asyncio 源代码，并确定这实际上是 asyncio本身。

我已将其提交给问题跟踪器 here对于那些有同样问题的人，并用我创建的解决方法回答了我自己的问题。

<小时/>

编辑:根据 ParkerD 的要求，以下是产生上述问题的完整可运行示例:

编辑 2:更新了示例以使用 Python 3.7+ 中的新 asyncio.run 和 asyncio.create_task 功能

import asyncio

async def coro():
    cond = asyncio.Condition()
    print("Taking lock...")
    async with cond:
        print("Lock acquired.")
        print("Waiting!")
        await asyncio.wait_for(cond.wait(), timeout=999)
        print("Was notified!")
    print("Lock released.")

async def cancel_after_5(c):
    task = asyncio.create_task(c)
    await asyncio.sleep(5)
    task.cancel()
    await asyncio.wait([task])

asyncio.run(cancel_after_5(coro()))

Answer 1

正如问题末尾所述，我已确定该问题实际上是库中的错误。我会重申该错误的问题跟踪器是 here ，并介绍我的解决方法。

以下函数基于 wait_for 本身(来源 here )，并且是专门用于等待条件的版本，并额外保证取消它是安全的。

调用 wait_on_condition_with_timeout(cond, timeout) 大致相当于 asyncio.wait_for(cond.wait(), timeout)。

async def wait_on_condition_with_timeout(condition: asyncio.Condition, timeout: float) -> bool:
    loop = asyncio.get_event_loop()

    # Create a future that will be triggered by either completion or timeout.
    waiter = loop.create_future()

    # Callback to trigger the future. The varargs are there to consume and void any arguments passed.
    # This allows the same callback to be used in loop.call_later and wait_task.add_done_callback,
    # which automatically passes the finished future in.
    def release_waiter(*_):
        if not waiter.done():
            waiter.set_result(None)

    # Set up the timeout
    timeout_handle = loop.call_later(timeout, release_waiter)

    # Launch the wait task
    wait_task = loop.create_task(condition.wait())
    wait_task.add_done_callback(release_waiter)

    try:
        await waiter  # Returns on wait complete or timeout
        if wait_task.done():
            return True
        else:
            raise asyncio.TimeoutError()

    except (asyncio.TimeoutError, asyncio.CancelledError):
        # If timeout or cancellation occur, clean up, cancel the wait, let it handle the cancellation,
        # then re-raise.
        wait_task.remove_done_callback(release_waiter)
        wait_task.cancel()
        await asyncio.wait([wait_task])
        raise

    finally:
        timeout_handle.cancel()

关键部分是，如果发生超时或取消，该方法会等待条件重新获取锁，然后重新引发异常:

except (asyncio.TimeoutError, asyncio.CancelledError):
        # If timeout or cancellation occur, clean up, cancel the wait, let it handle the cancellation,
        # then re-raise.
        wait_task.remove_done_callback(release_waiter)
        wait_task.cancel()
        await asyncio.wait([wait_task])  # This line is missing from the real wait_for
        raise

我已经在 Python 3.6.9 上对此进行了测试，它运行得很好。 3.7 和 3.8 中也存在同样的错误，所以我想它对这些版本也很有用。如果您想知道错误何时会得到修复，请查看上面的问题跟踪器。如果您想要除 Condition 之外的版本，则更改参数和 create_task 行应该很简单。