python - python-如何将C函数实现为可等待的(协程)

Question

环境:C和micropython虚拟机中的协作RTOS是任务之一。

为了使VM不阻止其他RTOS任务，我在 RTOS_sleep() 中插入vm.c:DISPATCH()，以便在执行每个字节码后，VM放弃对下一个RTOS任务的控制。

我创建了一个uPy接口(interface)，以使用生产者－消费者设计模式从物理数据总线(可能是CAN，SPI，以太网)异步获取数据。

在uPy中的用法:

can_q = CANbus.queue()
message = can_q.get()

在C中的实现是 can_q.get()不会阻止RTOS:它轮询C队列，如果未收到消息，它将调用 RTOS_sleep()使另一个任务有机会填充队列。事情是同步的，因为C队列仅由另一个RTOS任务更新，并且RTOS任务仅在调用 RTOS_sleep()即 合作时切换

C实现基本上是:

// gives chance for c-queue to be filled by other RTOS task
while(c_queue_empty() == true) RTOS_sleep(); 
return c_queue_get_message();

尽管Python语句 can_q.get()不会阻止RTOS，但它确实会阻止uPy脚本。
我想重写它，以便可以将它与 async def(即 协程)一起使用，并且不阻塞uPy脚本。

不知道 the syntax，但是像这样:

can_q = CANbus.queue()
message = await can_q.get()

问题

如何编写C函数，以便可以对其进行 await？

我更喜欢CPython和micropython的答案，但我会接受仅CPython的答案。

Answer 1

注意:此答案涵盖CPython和asyncio框架。但是，这些概念应适用于其他Python实现以及其他异步框架。

How do I write a C-function so I can await on it?

编写可以等待结果的C函数的最简单方法是让它返回一个已经制成的等待对象，例如 asyncio.Future 。在返回 Future之前，该代码必须安排通过某种异步机制设置将来的结果。所有这些基于协程的方法都假定您的程序在某个事件循环下运行，该事件循环知道如何调度协程。

但是返回 future 并不总是足够的-也许我们想定义一个具有任意数量的悬挂点的对象。返回 future 仅暂停一次(如果返回的 future 未完成)，一旦完成 future 就恢复，仅此而已。等效于包含多个 async def的 await的等待对象无法通过返回Future来实现，它必须实现协程通常实现的协议(protocol)。这有点像实现自定义 __next__的迭代器，并且可以代替生成器使用。

定义等待的自定义

要定义我们自己的等待类型，我们可以转到PEP 492，它可以将哪些 specifies确切地传递给 await。除了使用 async def定义的Python函数外，用户定义的类型还可以通过定义 __await__特殊方法使对象等待，该方法将Python / C映射到 tp_as_async.am_await结构的 PyTypeObject部分。

这意味着在Python / C中，您必须执行以下操作:

为扩展类型的 tp_as_async 字段指定非NULL值。

将其 am_await 成员指向一个C函数，该C函数接受您的类型的实例并返回实现iterator protocol的另一个扩展类型的实例，即，定义 tp_iter (通常定义为PyIter_Self)和 tp_iternext 。

迭代器的tp_iternext必须推进协程的状态机。 tp_iternext的每个非异常返回都对应一个暂停，并且最终StopIteration异常表示协程的最终返回。返回值存储在value的StopIteration属性中。

为了使协程有用，它还必须能够与驱动它的事件循环进行通信，以便它可以指定挂起后何时恢复。由asyncio定义的大多数协程都期望在asyncio事件循环下运行，并在内部使用 asyncio.get_event_loop()(和/或接受显式的 loop参数)来获取其服务。

协程示例

为了说明Python / C代码需要实现什么，让我们考虑用Python async def表示的简单协程，例如 asyncio.sleep()的等效形式:

async def my_sleep(n):
    loop = asyncio.get_event_loop()
    future = loop.create_future()
    loop.call_later(n, future.set_result, None)
    await future
    # we get back here after the timeout has elapsed, and
    # immediately return

my_sleep创建一个 Future ，安排它在n秒内完成(其结果被设置)，并暂停自身直到将来完成。最后一部分使用 await，其中 await x的意思是“允许 x决定我们现在将暂停还是继续执行”。不完整的将来总会决定暂停，而异步 Task协程驱动程序的特殊情况使 future 无限期地暂停它们，并将其完成与恢复任务联系起来。其他事件循环(curio等)的挂起机制在细节上可能有所不同，但是基本思想是相同的: await是可选的执行挂起。
__await__()返回一个生成器

要将其转换为C，我们必须摆脱神奇的 async def函数定义以及 await悬挂点。删除 async def非常简单:等效的普通函数只需要返回一个实现 __await__的对象:

def my_sleep(n):
    return _MySleep(n)

class _MySleep:
    def __init__(self, n):
        self.n = n

    def __await__(self):
        return _MySleepIter(self.n)

__await__操作符将自动调用 _MySleep返回的 my_sleep()对象的 await方法，以将等待对象(传递给 await的所有对象)转换为迭代器。该迭代器将用于询问等待的对象是选择暂停还是提供值。这很像 for o in x语句调用 x.__iter__()将可迭代的 x转换为具体迭代器的方式。

当返回的迭代器选择暂停时，只需要产生一个值即可。值的含义(如果有的话)将由协程驱动程序解释，通常是事件循环的一部分。当迭代器选择停止执行并从 await返回时，它需要停止迭代。使用生成器作为便利的迭代器实现， _MySleepIter看起来像这样:

def _MySleepIter(n):
    loop = asyncio.get_event_loop()
    future = loop.create_future()
    loop.call_later(n, future.set_result, None)
    # yield from future.__await__()
    for x in future.__await__():
        yield x

当 await x映射到 yield from x.__await__()时，我们的生成器必须用尽 future.__await__()返回的迭代器。否则，如果将来未完成，则 Future.__await__返回的迭代器将产生结果，并返回将来的结果(在此我们忽略，但 yield from实际上提供了结果)。

返回自定义迭代器的 __await__()
C语言中 my_sleep的C实现的最终障碍是 _MySleepIter的生成器的使用。幸运的是，任何生成器都可以转换为有状态的迭代器，该迭代器的 __next__将执行这段代码直到下一次等待或返回。 __next__实现生成器代码的状态机版本，其中 yield通过返回值表示， return通过引发 StopIteration表示。例如:

class _MySleepIter:
    def __init__(self, n):
        self.n = n
        self.state = 0

    def __iter__(self):  # an iterator has to define __iter__
        return self

    def __next__(self):
        if self.state == 0:
            loop = asyncio.get_event_loop()
            self.future = loop.create_future()
            loop.call_later(self.n, self.future.set_result, None)
            self.state = 1
        if self.state == 1:
            if not self.future.done():
                return next(iter(self.future))
            self.state = 2
        if self.state == 2:
            raise StopIteration
        raise AssertionError("invalid state")

翻译成C

上面有很多类型，但是可以使用，并且仅使用可以通过本机Python / C函数定义的构造。

实际上，将这两个类转换为C非常简单，但超出了此答案的范围。