如何使用asyncio.Python中任务中的条件<v3.10

Question

不能使用asyncio的任何原因。任务内的条件?

c = asyncio.Condition()
async def a():
print("A ..")
# await asyncio.sleep(0.2) # This works
async with c:
# RuntimeError: Task <Task pending coro=<a() running at ..this file..:13>> got Future <Future pending> attached to a different loop
await c.wait()   #
async def main():
asyncio.get_event_loop().create_task(a())
await asyncio.sleep(2)

显示:"Got Future附加到另一个循环">

我不认为我创建了一个新的循环。

完整的例子在这里:

import asyncio
c = asyncio.Condition()
async def a():
print("A ..")
# await asyncio.sleep(0.2) # This works
async with c:
# RuntimeError: Task <Task pending coro=<a() running at ..this file..:13>> got Future <Future pending> attached to a different loop
await c.wait()   #
print("A done")
async def b():
await asyncio.sleep(2)
print("B ..")
async with c:
c.notify_all() 
print("B done")
await asyncio.sleep(1)
async def main():
asyncio.get_event_loop().create_task(a())
await b()
asyncio.run(main())

我在Python 3.7中看到同样的错误。3.8, 3.9 .

Answer 1

asyncio.Condition.notify_all()状态的文档:

锁必须在调用此方法之前获得，并在调用后不久释放。如果使用未锁定的锁调用，则会引发RuntimeError错误。

调用c.wait()时a中的锁被释放，因此调用c.notify_all()时c中的Lock被解锁。

在调用notify_all()之前需要持有锁。使用

async with c:
c.notify_all()

使您的示例按预期工作。

更新我只在Python 3.10.1上测试了它，它是这样工作的。事实上，当我在Python 3.8.5上运行它时，它失败了。但是这里的问题源于将Condition用作全局变量。在您的示例中，Condition是在创建事件循环之前创建的，因此我假设它没有正确地附加到稍后创建的事件循环。我以某种方式更新了您的示例，即Condition是用运行循环创建的。这使得该示例在Python 3.8.5中再次工作:

import asyncio
async def a(c):
print("A ..")
async with c:
await c.wait()
print("A done")
async def b(c):
await asyncio.sleep(2)
print("B ..")
async with c:
c.notify_all()
print("B done")
await asyncio.sleep(1)
async def main():
c = asyncio.Condition() # loop already running
asyncio.create_task(a(c)) # get_event_loop() also works but the use is discouraged in the docs
await b(c)
asyncio.run(main())

Answer 2

正如@JanWilamowski指出的，这似乎是Python的一个bug <v3.10。>

Booo !

似乎你也不能通过从条件切换到队列来解决这个问题(见下文)-我猜队列在内部使用条件。Simlarly……

async def main():
loop.create_task(a())
...
loop = asyncio.new_event_loop()
loop.run_until_complete(main())
# asyncio.run(main())

然而，由于不清楚的原因，这个确实工作:

async def main():
loop.create_task(a())
...
loop = asyncio.get_event_loop()   // @@@@
loop.run_until_complete(main())

完整的工作示例如下:

import asyncio
import time
USE_QUEUE = False
if not USE_QUEUE:
c = asyncio.Condition()
else:
q = asyncio.Queue()
async def a():
print("A ..")
# await asyncio.sleep(0.2) # This works
if not USE_QUEUE:
async with c:
await c.wait()
else:
result = await q.get()
q.task_done()
print("result", result)
print("A done", time.time())
async def b():
await asyncio.sleep(1)
print("B ..", time.time())
if not USE_QUEUE:
async with c:
c.notify_all()
else:
result = await q.put(123)
await asyncio.sleep(1)
print("B done")
async def main():
loop.create_task(a())
# asyncio.get_event_loop().create_task(a())
await b()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())