编写中断处理程序

在合适的硬件上,MicroPython 提供了使用 Python 编写中断处理程序的能力。中断处理程序——也称为中断服务例程(ISR)——以回调函数的形式定义。它们会在响应某个事件(例如定时器触发或引脚上的电压变化)时执行。此类事件可能发生在程序代码执行过程中的任意时刻。这会带来重大影响,其中一些是 MicroPython 语言所特有的,另一些则是所有能够响应实时事件的系统所共有的。本文档首先介绍语言特有的问题,然后为初学者简要介绍实时编程。

本介绍使用了诸如“慢”或“尽可能快”之类的模糊术语。这是有意为之的,因为速度取决于具体应用。ISR 可接受的执行时长取决于中断发生的速率、主程序的性质以及是否存在其他并发事件。

MicroPython 相关问题

紧急异常缓冲区

如果 ISR 中发生错误,除非为此目的创建了一个特殊缓冲区,否则 MicroPython 将无法生成错误报告。如果在任何使用中断的程序中包含以下代码,调试将会变得更简单。

import micropython

micropython.alloc_emergency_exception_buf(100)

紧急异常缓冲区只能容纳一个异常堆栈跟踪。这意味着,如果在堆被锁定期间处理某个异常的过程中抛出了第二个异常,那么第二个异常的堆栈跟踪将替换原来的——即使第二个异常被干净地处理掉了也是如此。如果稍后打印该缓冲区,这可能会导致令人困惑的异常消息。

简单性

出于种种原因,让 ISR 代码尽可能简短而简单是非常重要的。它应该只执行在引发它的事件发生后必须立即完成的工作:可以推迟的操作应交给主程序循环来处理。通常,ISR 会处理引发中断的硬件设备,使其为下一次中断做好准备。它会通过更新共享数据来与主循环通信,以表明中断已经发生,然后返回。ISR 应尽快将控制权交还给主循环。这不是 MicroPython 特有的问题,因此将在 下文 中更详细地介绍。

ISR 与主程序之间的通信

ISR 通常需要与主程序通信。实现这一点最简单的方法是通过一个或多个共享数据对象,这些对象既可以声明为全局变量,也可以通过类来共享(参见下文)。这样做存在各种限制和隐患,下文将更详细地介绍。整数、bytesbytearray 对象通常用于此目的,此外还有可以存储各种数据类型的数组(来自 array 模块)。

将对象方法用作回调

MicroPython 支持这一强大的技术,它使 ISR 能够与底层代码共享实例变量。它还使实现设备驱动程序的类能够支持多个设备实例。以下示例使两个 LED 以不同的速率闪烁。

import machine
import micropython

micropython.alloc_emergency_exception_buf(100)


class Foo(object):
    def __init__(self, freq, led):
        self.led = led
        self.timer = machine.Timer(-1, freq=freq, callback=self.cb, hard=True)

    def cb(self, tim):
        self.led.toggle()


red = Foo(1, machine.LED("LED_RED"))
green = Foo(0.8, machine.LED("LED_GREEN"))

在此示例中,red 实例由一个 1 Hz 的虚拟定时器驱动红色 LED:每次定时器触发时都会调用 red.cb(),从而切换红色 LED。green 实例的工作方式类似,由一个 0.8 Hz 的定时器切换绿色 LED。使用实例方法带来两个好处。首先,单个类使代码能够在多个硬件实例之间共享。其次,作为绑定方法,回调函数的第一个参数是 self。这使回调能够访问实例数据,并在连续调用之间保存状态。例如,如果上述类有一个在构造函数中设为零的变量 self.count,则 cb() 可以递增该计数器。这样,redgreen 实例就会各自独立地维护每个 LED 改变状态的次数。

创建 Python 对象

ISR 不能创建 Python 对象的实例。这是因为 MicroPython 需要从称为 heap 的空闲内存块存储区为对象分配内存。在中断处理程序中不允许这样做,因为堆分配不是可重入的。换句话说,中断可能在主程序正在执行分配操作的中途发生——为了维护堆的完整性,解释器禁止在 ISR 代码中进行内存分配。

由此带来的一个后果是,ISR 不能使用浮点运算;这是因为浮点数是 Python 对象。同样,ISR 也不能向列表追加元素。在实践中,可能很难准确判断哪些代码结构会尝试执行内存分配并引发错误消息:这也是让 ISR 代码简短而简单的另一个原因。

避免此问题的一种方法是让 ISR 使用预先分配的缓冲区。例如,类的构造函数创建一个 bytearray 实例和一个布尔标志。ISR 方法将数据赋值到缓冲区中的相应位置并设置该标志。内存分配发生在对象被实例化时的主程序代码中,而不是发生在 ISR 中。

MicroPython 库的 I/O 方法通常提供使用预先分配缓冲区的选项。例如,machine.I2C.readfrom_into() 会读取数据到调用方提供的可变缓冲区中:这使其能够在 ISR 中使用。

一种无需使用类或全局变量来创建对象的方法如下:

def set_volume(t, buf=bytearray(3)):
    buf[0] = 0xa5
    buf[1] = t >> 4
    buf[2] = 0x5a
    return buf

编译器会在函数首次加载时(通常是在导入函数所在的模块时)实例化默认的 buf 参数。

创建对绑定方法的引用时也会发生一次对象创建。这意味着 ISR 不能将绑定方法传递给函数。一种解决方案是在类的构造函数中创建对该绑定方法的引用,然后在 ISR 中传递该引用。例如:

class Foo():
    def __init__(self):
        self.bar_ref = self.bar  # Allocation occurs here
        self.x = 0.1
        self.tim = machine.Timer(-1, freq=2, callback=self.cb, hard=True)

    def bar(self, _):
        self.x *= 1.2
        print(self.x)

    def cb(self, t):
        # Passing self.bar would cause allocation.
        micropython.schedule(self.bar_ref, 0)

其他技巧包括在构造函数中定义并实例化该方法,或者以参数 self 来传递 Foo.bar()

Python 对象的使用

由于 Python 的工作方式,对象还受到进一步的限制。当执行 import 语句时,Python 代码会被编译为 bytecode,通常一行代码会映射为多个字节码。代码运行时,解释器读取每个字节码并将其作为一系列机器码指令来执行。鉴于中断可能在机器码指令之间的任意时刻发生,原始的那行 Python 代码可能只被部分执行。因此,在主循环中被修改的 Python 对象(例如集合、列表或字典)在中断发生的那一刻可能缺乏内部一致性。

典型的结果如下。在极少数情况下,ISR 会恰好在对象被部分更新的那一刻运行。当 ISR 尝试读取该对象时,就会发生崩溃。由于此类问题通常只在罕见、随机的情况下出现,它们可能很难诊断。有一些方法可以规避此问题,将在下文的 临界区 中介绍。

明确什么构成对对象的修改是很重要的。改变数组或字节数组的内容是安全的。这是因为字节或字是通过单条不可中断的机器码指令写入的:用实时编程的术语来说,这种写入是原子的。更新字典项也是如此,因为字典项是机器字,即整数或指向对象的指针。用户自定义对象可能会实例化一个数组或字节数组。主循环和 ISR 都可以改变这些内容,这是有效的。

当对象的结构被改变时就会产生隐患,尤其是在字典的情况下。添加或删除键可能会触发重新哈希。如果硬 ISR 在重新哈希进行过程中运行并尝试访问某一项,就可能发生崩溃。在内部,全局变量是以字典的形式实现的。因此,主程序应在启动会产生硬中断的过程之前创建好所有必需的全局变量。应用程序代码还应避免删除全局变量。

MicroPython 支持任意精度的整数。介于 230 -1 和 -230 之间的值将存储在单个机器字中。更大的值则存储为 Python 对象。因此,对长整数的更改不能视为原子的。在 ISR 中使用长整数是不安全的,因为在变量值改变时可能会尝试进行内存分配。

克服浮点数限制

一般来说,最好避免在 ISR 代码中使用浮点数:硬件设备通常处理整数,向浮点数的转换通常在主循环中完成。然而,有少数 DSP 算法需要浮点运算。在具有硬件浮点的平台上(例如基于 STM32 的 OpenMV Cam),可以使用内联 ARM Thumb 汇编来绕过此限制。这是因为处理器将浮点值存储在一个机器字中;这些值可以通过浮点数组在 ISR 与主程序代码之间共享。

使用 micropython.schedule

此函数使 ISR 能够调度一个回调,使其“很快”执行。该回调被排入队列,等待在堆未被锁定时执行。因此,它可以创建 Python 对象并使用浮点数。该回调还保证在主程序已完成对 Python 对象的任何更新之时运行,因此回调不会遇到被部分更新的对象。

典型用法是处理传感器硬件。ISR 从硬件采集数据,并使其能够发出进一步的中断。然后它调度一个回调来处理这些数据。

被调度的回调应遵循下文概述的中断处理程序设计原则。这是为了避免由 I/O 活动以及共享数据的修改所导致的问题,这些问题可能出现在任何抢占主程序循环的代码中。

需要结合中断可能发生的频率来考虑执行时间。如果在前一个回调执行期间发生中断,则会有该回调的另一个实例被排入队列等待执行;它将在当前实例完成后运行。因此,持续的高中断重复率会带来队列无限制增长并最终以 RuntimeError 失败的风险。

如果要传递给 schedule() 的回调是一个绑定方法,请参考“创建 Python 对象”中的注意事项。

异常

如果 ISR 引发异常,该异常不会传播到主循环。除非该异常被 ISR 代码处理,否则中断将被禁用。

与 asyncio 对接

当 ISR 运行时,它可能会抢占 asyncio 调度器。如果 ISR 执行了 asyncio 操作,调度器的运行可能会被打乱。无论中断是硬中断还是软中断,这都适用;如果 ISR 已通过 micropython.schedule 将执行权移交给另一个函数,这同样适用。特别是,在 ISR 上下文中创建或取消任务是无效的。与 asyncio 交互的安全方法是实现一个协程,并通过 asyncio.ThreadSafeFlag 来进行同步。以下片段演示了如何响应中断来创建任务:

tsf = asyncio.ThreadSafeFlag()


def isr(_):  # Interrupt handler
    tsf.set()


async def foo():
    while True:
        await tsf.wait()
        asyncio.create_task(bar())

在此示例中,从 ISR 的执行到 foo() 的执行之间会存在一段可变的延迟。这是协作式调度所固有的。最大延迟取决于具体应用和平台,但通常可以以数十毫秒计。

一般性问题

这仅仅是对实时编程这一主题的简要介绍。初学者应注意,实时程序中的设计错误可能导致特别难以诊断的故障。这是因为它们可能很少出现,而且出现的间隔基本上是随机的。最初就把设计做对,并在问题出现之前预见到它们,这一点至关重要。中断处理程序和主程序都需要在充分理解以下问题的基础上进行设计。

中断处理程序设计

如上所述,ISR 的设计应尽可能简单。它们应始终在一段简短、可预测的时间内返回。这一点很重要,因为当 ISR 正在运行时,主循环并未运行:主循环不可避免地会在代码中的随机位置出现执行暂停。此类暂停可能成为难以诊断的 bug 的来源,尤其是当其持续时间较长或不固定时。为了理解 ISR 运行时间的影响,需要对中断优先级有基本的了解。

中断是按照优先级方案来组织的。ISR 代码本身可能被更高优先级的中断所打断。如果两个中断共享数据,这就会带来影响(参见下文“临界区”一节)。如果发生这样的中断,它会向 ISR 代码中插入一段延迟。如果在 ISR 运行期间发生较低优先级的中断,它将被延迟到 ISR 完成之后:如果延迟过长,较低优先级的中断可能会失败。慢速 ISR 的另一个问题是,在其执行期间发生了同类型的第二个中断。第二个中断将在第一个中断结束后被处理。然而,如果传入中断的速率持续超过 ISR 服务它们的能力,结果将不会令人愉快。

因此,应避免或尽量减少循环结构。通常应避免对引发中断的设备以外的设备进行 I/O:诸如磁盘访问、print 语句和 UART 访问之类的 I/O 相对较慢,其持续时间可能各不相同。这里还有一个问题是文件系统函数不是可重入的:在 ISR 和主程序中同时使用文件系统 I/O 将很危险。至关重要的是,ISR 代码不应等待某个事件。如果代码可以保证在可预测的时间内返回,例如切换某个引脚或 LED,则 I/O 是可以接受的。通过 I2C 或 SPI 访问引发中断的设备可能是必要的,但应计算或测量此类访问所需的时间,并评估其对应用的影响。

通常需要在 ISR 与主循环之间共享数据。这可以通过全局变量,或通过类变量或实例变量来实现。变量通常是整数或布尔类型,或者整数或字节数组(预先分配的整数数组比列表提供更快的访问速度)。当 ISR 修改多个值时,需要考虑这样一种情况:中断恰好在主程序已访问了部分而非全部值的时候发生。这可能导致不一致。

考虑以下设计。ISR 将传入的数据存储到一个字节数组中,然后将接收到的字节数加到一个表示已准备好待处理的总字节数的整数上。主程序读取该字节数,处理这些字节,然后将待处理字节数清零。这一切都能正常工作,直到中断恰好在主程序刚读取完字节数之后发生。ISR 将新增的数据放入缓冲区并更新接收到的数量,但主程序已经读取了原来的数量,因此只处理最初接收到的数据。新到达的字节就丢失了。

有多种方法可以避免这种隐患,最简单的是使用循环缓冲区。如果无法使用具有固有线程安全性的结构,下文还介绍了其他方法。

可重入性

如果某个函数或方法在主程序与一个或多个 ISR 之间,或在多个 ISR 之间共享,可能会产生潜在的隐患。这里的问题在于,该函数本身可能被中断,从而运行该函数的另一个实例。如果会发生这种情况,则该函数必须设计为可重入的。如何做到这一点是一个超出本教程范围的高级主题。

临界区

临界区代码的一个例子是访问多个可能受 ISR 影响的变量的代码。如果中断恰好在对各个变量的访问之间发生,它们的值就会不一致。这是一种被称为竞态条件的隐患的实例:ISR 与主程序循环争相改变这些变量。为避免不一致,必须采用某种手段来确保在临界区持续期间 ISR 不会改变这些值。实现这一点的一种方法是在临界区开始之前调用 machine.disable_irq(),并在结束时调用 machine.enable_irq()。以下是这种方法的一个示例:

import machine
import micropython
import array
import random
import time

micropython.alloc_emergency_exception_buf(100)


class BoundsException(Exception):
    pass


ARRAYSIZE = const(20)
index = 0
data = array.array('i', [0] * ARRAYSIZE)


def callback1(t):
    global data, index
    for x in range(5):
        data[index] = random.getrandbits(30)  # simulate input
        index += 1
        if index >= ARRAYSIZE:
            raise BoundsException('Array bounds exceeded')


tim = machine.Timer(-1, freq=100, callback=callback1, hard=True)

for loop in range(1000):
    if index > 0:
        irq_state = machine.disable_irq()  # Start of critical section
        for x in range(index):
            print(data[x])
        index = 0
        machine.enable_irq(irq_state)  # End of critical section
        print('loop {}'.format(loop))
    time.sleep_ms(1)

tim.deinit()

临界区可以仅由一行代码和一个变量组成。考虑以下代码片段。

count = 0


def cb(): # An interrupt callback
    count += 1


def main():
    # Code to set up the interrupt callback omitted
    while True:
        count += 1

此示例阐释了一种隐蔽的 bug 来源。主循环中的 count += 1 这一行带有一种被称为“读取-修改-写入”的特定竞态条件隐患。这是实时系统中 bug 的经典成因。在主循环中,MicroPython 读取 count 的值,给它加 1,然后将其写回。在极少数情况下,中断恰好在读取之后、写入之前发生。中断修改了 count,但当 ISR 返回时,它的更改会被主循环覆盖。在实际系统中,这可能导致罕见、不可预测的故障。

如上所述,如果在主代码中修改了某个 Python 内置类型的实例,并且该实例在 ISR 中被访问,则应格外小心。执行修改的代码应被视为临界区,以确保当 ISR 运行时该实例处于有效状态。

如果在不同的 ISR 之间共享某个数据集,则需要格外小心。这里的隐患在于,较高优先级的中断可能在较低优先级的中断已部分更新共享数据时发生。处理这种情况是一个超出本介绍范围的高级主题,这里仅指出下文描述的互斥锁对象有时可以派上用场。

在临界区持续期间禁用中断是通常且最简单的做法,但它会禁用所有中断,而不仅仅是那个有可能引发问题的中断。长时间禁用某个中断通常是不可取的。对于定时器中断,这会给回调发生的时间引入变动。对于设备中断,这可能导致设备被服务得太晚,从而可能造成数据丢失或设备硬件中的溢出错误。与 ISR 一样,主代码中的临界区也应具有简短、可预测的持续时间。

处理临界区的一种能从根本上减少中断被禁用时间的方法是使用一种称为互斥锁(mutex,名称源自“互斥”这一概念)的对象。主程序在运行临界区之前锁定互斥锁,并在结束时解锁它。ISR 则测试互斥锁是否被锁定。如果被锁定,它就避开临界区并返回。设计上的挑战在于定义当对临界变量的访问被拒绝时 ISR 应该做什么。一个简单的互斥锁示例可以在 这里 找到。请注意,互斥锁代码确实会禁用中断,但仅持续八条机器指令的时间:这种方法的好处是其他中断几乎不受影响。

中断与 REPL

中断处理程序(例如与定时器相关联的那些)在程序终止后仍可能继续运行。在你本以为引发回调的对象应已超出作用域的情况下,这可能产生意料之外的结果。例如,在 OpenMV Cam 上:

def bar():
    foo = machine.Timer(-1, freq=4, callback=lambda t: print('.', end=''), hard=True)

bar()

这会一直运行,直到定时器被显式禁用,或者使用 Ctrl-D 复位开发板为止。