1

我在尝试制作调用异步函数的同步函数时遇到了一些问题。（python 3.6.9，cocotb 1.4.0）

如以下示例代码所示。该read_cb函数将调用read函数（在FakeDriver类中）。

运行后，我得到错误

yield self._fake_lock()
RuntimeError: Task got bad yield: <cocotb.decorators.RunningCoroutine object at 0x7f7fecdbfe10>

我想要的是

init FakerDriver
locking... 
locking done
read...
addr: 0x01 
unlocking... 
unlocking done
read done

import cocotb
import asyncio
from cocotb.decorators import coroutine
from cocotb.triggers import Event

class FakeDriver():
    def __init__(self):
        print("init FakeDriver")
        self.busy_event = Event("driver_busy")
        self.busy = False

    @coroutine
    def read(self, addr):
        print("read...")

        yield self._fake_lock()
        print("addr: ", addr)
        self._fake_unlock()

        print("read done")

    @coroutine
    def _fake_lock(self):
        print("locking...")
        if self.busy:
            yield self.busy_event.wait()

        self.busy_event.clear()
        self.busy = True
        print("locking done")

    def _fake_unlock(self):
        print("unlocking...")
        self.busy = False
        self.busy_event.set()
        print("unlocking done")

def read_cb():
    dri = FakeDriver()
    loop = asyncio.get_event_loop()
    task = loop.create_task(dri.read("0x01"))
    ret = loop.run_until_complete(task)
    loop.close()

if __name__ == "__main__":
    read_cb()
4

1 回答 1

1

不要将 cocotb 自己的协程实现与 asyncio 混为一谈。在您的情况下，完全摆脱asyncio导入，并使用旧的 cocotb 1.4.0，使用 cocotbfork()而不是create_task()（如https://docs.cocotb.org/en/v1.4.0/quickstart.html?highlight=中所述fork#parallel-and-sequential-execution）。

在使用yieldcocotb 中已弃用的等创建大量新代码之前，请考虑将 cocotb 升级到 1.6.1 并使用async def/ await（同样来自cocotb, not asyncio），而不是fork(), 使用start_soon()（参见https://docs.cocotb.org/en /v1.6.1/coroutines.html#concurrent-execution和https://www.fossi-foundation.org/2021/10/20/cocotb-1-6-0）。

（我现在看到你也在https://github.com/cocotb/cocotb/issues/2819中问过这个问题）

于 2021-12-29T12:36:44.027 回答