使用 cocotb 在列表中生成协程

yielding a coroutine in a list with cocotb

我有一个等待事件设置的协程:

@cocotb.coroutine
def wb_RXDR_read(self):
    """ waiting util RXDR is read """
    if not self._RXDR_read_flag:
        while True:
            yield self._RXDR_read_event.wait()
            break

我想在超时的情况下“屈服”。然后我这样做了:

        RXDR_timeout = Timer(250, units="us")
        ret = yield [RXDR_timeout, self.wb_RXDR_read()]
        if ret == RXDR_timeout:
            self._dut._log.error("Timeout on waiting RXDR to be read")
            raise TestError()

但是我得到这个错误:

2ns ERROR    Coroutine i2c_write yielded something the scheduler can't handle
                      Got type: <type 'list'> repr: [<cocotb.triggers.Timer object at 0x7f2098cb1350>, <cocotb.decorators.RunningCoroutine object at 0x7f2098cb1610>] str: [<cocotb.triggers.Timer object at 0x7f2098cb1350>, <cocotb.decorators.RunningCoroutine object at 0x7f2098cb1610>]
                      Did you forget to decorate with @cocotb.coroutine?

我的协程是用@cocotb.coroutine装饰的。如果我单独 yield 它有效:

yield self.wb_RXDR_read() # <- that works

但我无法将其放入列表中。是否可以将协程放入列表中以像 unix select() 那样进行阻塞?还是保留给 Trigger class ?

好的,我找到了解决方案。事实上协程不能像时尚本身那样在 select 中触发。它应该首先作为一个线程启动,并且要检测协程的结束,.join() 方法必须放在 yield 列表中:

    RXDR_timeout = Timer(250, units="us")
    RXDR_readth = cocotb.fork(self.wb_RXDR_read())
    ret = yield [RXDR_timeout, RXDR_readth.join()]
    if ret == RXDR_timeout:
        self._dut._log.error("Timeout on waiting RXDR to be read")
        raise TestError()

要记住的是:

  • 我们可以产生一个协程
  • 要生成多个协同程序,我们必须 fork()join()