python如何将Cfunction实现为可等待（coroutine）

Question

环境：C和micropython虚拟机的协同RTOS是其中的一个任务。在

为了使VM不阻塞其他RTOS任务，我在^{}中插入RTOS_sleep()，以便在执行每个字节码之后，VM将控制权交给下一个RTOS任务。在

我创建了一个uPy接口，使用producer-consumer设计模式从物理数据总线（可以是CAN、SPI、ethernet）异步获取数据。在

在纽约大学的用法：

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

在C中的实现使得can_q.get()不阻塞RTOS：它轮询一个C队列，如果没有收到消息，它调用RTOS_sleep()给另一个任务填充队列的机会。事情是同步的，因为C队列只由另一个RTOS任务更新，而RTOS任务只在调用RTOS_sleep()时切换，即协作

C实现基本上是：

^{pr2}$

尽管Python语句can_q.get()没有阻止rto，但是它确实阻止了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函数的最简单方法是让它返回一个已经生成的可等待对象，例如^{}。在返回Future之前，代码必须安排由某个异步机制设置将来的结果。所有这些基于协程的方法都假设您的程序在某个事件循环下运行，该循环知道如何调度协同程序。在

但是返回一个未来并不总是足够的-也许我们想定义一个具有任意数量的悬挂点的对象。返回一个future只会暂停一次（如果返回的future不完整），一旦future完成就继续，就这样。与包含多个async def的async def等价的可等待对象不能通过返回future来实现，它必须实现协同路由通常实现的协议。这有点像一个实现自定义__next__的迭代器，并被用来代替生成器。在

定义可等待的自定义

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

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

• 为扩展类型的^{}字段指定一个非空值。在
• 让它的^{}成员指向一个C函数，该函数接受您的类型的实例，并返回另一个实现iterator protocol的扩展类型的实例，即定义^{}（通常定义为PyIter_Self）和{a7}。在
• 迭代器的tp_iternext必须推进协程的状态机。来自tp_iternext的每个非异常返回都对应于一个挂起，而最终的StopIteration异常表示从协程的最终返回。返回值存储在StopIteration的value属性中。在

为了使协程有用，它还必须能够与驱动它的事件循环通信，以便它可以指定在挂起后何时恢复。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创建一个^{}，安排它在n秒内完成（其结果变为设置），并将自身挂起，直到将来完成。最后一部分使用await，其中await x表示“允许x来决定我们现在是暂停还是继续执行”。不完整的未来总是决定挂起，而asyncioTask协程驱动程序的特殊情况产生了无限期暂停它们的期货，并将它们的完成与任务的恢复联系起来。其他事件循环（curio等）的挂起机制可能在细节上有所不同，但基本思想是相同的：await是可选的执行暂停。在

__await__()返回生成器

要将其转换为C语言，我们必须去掉神奇的async def函数定义，以及await暂停点。删除async def相当简单：等效的普通函数只需返回实现__await__的对象：

^{pr2}$

my_sleep()返回的_MySleep对象的__await__方法将由await运算符自动调用，以将一个可等待的对象（传递给await的任何对象）转换为iter阿托尔。这个迭代器将用来询问等待的对象是选择挂起还是提供一个值。这很像for o in x语句调用x.__iter__()将iterablex转换为具体的迭代器。在

当返回的迭代器选择挂起时，它只需要生成一个值。值的含义（如果有的话）将由协同程序驱动程序解释，通常是事件循环的一部分。当迭代器选择停止执行并从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不完整，Future.__await__返回的迭代器将生成future，否则返回future的结果（这里我们忽略了这个结果，但是{}实际上提供了这个结果）。在

__await__()返回自定义迭代器

在C中实现my_sleep的最后一个障碍是使用_MySleepIter的生成器。幸运的是，任何生成器都可以转换为有状态迭代器，它的__next__执行代码片段，直到下一个等待或返回。__next__实现了生成器代码的状态机版本，其中yield通过返回一个值来表示，return通过提升{}来表示。例如：

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相当简单，但超出了这个答案的范围。在