本文会简单介绍中断的基本原理,对MicroPython在处理中断时的一些机制和问题进行阐述,并对实时控制中断编程做简单的介绍。
中断是计算机系统中非常重要的一种机制,简单的说就是当有I/O或其它因素发出中断信号后,CPU会暂定执行当前代码保护好现场,然后执行中断处理程序。CPU执行完中断处理程序后,返回原来的执行点,恢复现场,继续执行原来的代码。
MicroPython也支持在合适的硬件上(与硬件和固件有相关性)编写中断处理程序。中断处理程序(也称为中断服务程序ISR),通常被定义为回调函数。当定时器或引脚电平变化等事件触发中断后,中断处理程序将被执行。
通常在需要对外部特定事件进行响应时需要用到中断,例如:
外部输入信号变化:当外部设备或传感器产生信号变化时,可能需要中断CPU当前的工作,转而去处理这个外部事件。
定时器中断:在特定的时间点或间隔,定时器可能产生中断,请求CPU执行相应的任务。
串口通信:在进行串行数据通信时,数据到达或特定事件发生(如接收到数据或发生错误)时,可能会触发串口中断。
ADC/DAC中断:在模拟数字转换或数字模拟转换过程中,当转换完成或出现错误时,可能会需要中断来处理。
代码要尽量简短。中断处理程序的代码应尽可能的短小精悍,避免长时间占用CPU,特别是在一些实时系统。
避免内存分配,在中断处理程序中不追加列表或插入字典,不要创建对象,不要使用浮点数。
可以使用 micropython.schedule 机制来处理上述限制。
当 ISR 返回多个字节时,使用预先分配的字节数组。如果要在 ISR 和主程序之间共享多个整数,建议使用数组 (array.array)。
在主程序和 ISR 之间共享数据时,应考虑在主程序访问数据前禁用中断,并在访问后立即重新启用中断(可以参考临界区)。
记得分配一个紧急异常缓冲区。
在MicroPython中,如果没有给中断处理程序分配异常缓冲区,中断处理程序中产生的异常堆栈将无法捕获,也就获取ISR产生的异常信息。可以通过为ISR分配紧急异常缓冲区的方式解决这一问题,这在调试的时候会比较有用。当然,如果确信ISR代码不会产生异常,也可以不用分配这个缓冲区,否则不建议这么做。分配缓冲区的代码非常简单:
import micropython
micropython.alloc_emergency_exception_buf(100)
紧急异常缓冲区只能容纳一个异常堆栈信息。这意味着 ISR 产生多个异常的情况下,后面的会覆盖之前的,也就是说只会保留最后一个。所以,当有多个异常产生时,打印出的异常信息可能并不是真正的问题所在。
ISR 代码应尽可能短小精悍。ISR 只应执行必须执行的操作,可以推迟执行的操作应委托给主程序执行。通常情况下,ISR 会处理导致中断的硬件设备,使其为下一次中断做好准备。它会通过更新共享数据与主程序循环通信,表明中断已经发生,然后返回。ISR 应当尽快将控制权交换主程序。
ISR通常需要与主程序通信。最简单的方法是通过一个或多个共享数据对象,这些对象可以声明为全局对象,也可以通过类共享的方式(见下文),这样做有一些限制和危险,后面会详细介绍。整数、bytes
和bytearray
以及数组对象(array.array)通常被用于共享数据,数组对象可以存储各种数据类型。
MicroPython支持ISR与底层代码共享对象实例,甚至可以实现让设备驱动程序类支持多个设备实例。下面的示例展示了使用定时器中断,让两个LED以不同的速率闪烁。
import pyb, micropython
micropython.alloc_emergency_exception_buf(100) #申请一个紧急异常缓冲区
class Foo(object):
def __init__(self, timer, led):
self.led = led
timer.callback(self.cb) #传递给Timer对象方法
def cb(self, tim): #用于回调的方法
self.led.toggle()
red = Foo(pyb.Timer(4, freq=1), pyb.LED(1))
green = Foo(pyb.Timer(2, freq=0.8), pyb.LED(2))
在这个例子中,分别创建了两个Foo的实例,一个是red,将定时器4与LED1关联,频率为1,另一个是green,将定时器2与LED2关联,频率为0.8。当定时器4中断发生时,red.cb()
被调用,LED1状态改变;同样,当定时器2中断时,执行green.cb(),LED2状态改变。使用实例方法有两个好处:首先,单个类可以在多个硬件实例之间共享代码。其次,作为一个绑定方法,回调函数的第一个参数是 self。这样,回调函数就能访问实例数据,并在连续调用之间保存状态。例如,如果上面的类在构造函数中将变量 self.count 设为零,cb() 就可以递增计数器。这样,红色和绿色实例就会保持各自独立的计数,记录每个 LED 改变状态的次数。
在ISR代码中不允许创建Python对象的实例。这是因为 MicroPython需要从名为堆(heap)的空闲内存块中为对象分配内存。这在中断处理程序中是不允许的,因为堆分配不是可重入的。换句话说,中断可能会在主程序完成堆分配的过程中发生–为了维护堆的完整性,解释器不允许在 ISR 代码中进行内存分配。
因为不允许在ISR代码中分配内存,所以会带来一系列的问题,包括:
可以看出,要解决内存问题的途径就是避免在ISR代码中分配内存,解决问题的方法之一是让ISR使用预分配的缓冲区。例如,一个类的构造函数会创建一个bytearray
实例和一个布尔标志。ISR代码可以把数据放到bytearray
并设置标志,这样内存分配发生在实例化对象时主程序而不是ISR中。
MicroPython库的I/O方法通常会提供使用预分配缓冲区的选项,如:pyb.i2c.recv()
可以接受一个可变缓冲区作为其第一个参数,这样就可以在ISR中使用它了。
这也就意味着不使用类或全局变量也可以创建对象,如:
def set_volume(t, buf=bytearray(3)): #加载模块时实例化默认的buf参数,注意不要长期持有buf
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 #此时分配内存
self.x = 0.1
tim = pyb.Timer(4)
tim.init(freq=2)
tim.callback(self.cb)
def bar(self, _):
self.x *= 1.2
print(self.x)
def cb(self, t):
#传递self.bar会导致内存分配,此处传递其引用
micropython.schedule(self.bar_ref, 0)
其他方法是在构造函数中定义并实例化该方法,或将Foo.bar()与参数self一起传递。
由于Python的工作方式,对象的使用会受到进一步的限制。当执行导入语句时,Python代码会被编译成字节码,一行代码通常会映射成多个字节码。代码运行时,解释器会读取每个字节码,并将其作为一系列机器码指令执行。由于中断可能随时发生在机器码指令之间,Python原始代码可能只执行了一部分,因此,在主循环中修改的Python对象(如集合、列表或字典)在中断发生时可能缺乏内部一致性。
典型的场景是,在极少数情况下ISR会在对象修改的时刻运行,当ISR试图读取对象时,就会导致崩溃。由于此类问题通常发生在罕见的随机场合,因此很难诊断。有一些方法可以规避这一问题,后面的 "临界区"不分会介绍。
所以需要清楚的界定什么是对象的修改,修改字典等内置类型是有问题的,而修改数组或字节数组的内容则没有问题,这是因为字节或字是一条机器码指令写入的,不能中断,用实时编程的术语来说就是写入是原子性的,而字典、列表等复杂对象的操作不具有原子性。用户自定义对象可以实例化整数、数组或字节数组,这些内容在主循环和 ISR 均可更改。
MicroPython支持任意精度的整数。介于-2^30 到 2^30-1之间的数值将存储在一个机器字中。更大的数值将存储为 Python对象。因此,对长整数的更改不能视为原子更改。在ISR中使用长整数是不安全的,因为当变量的值发生变化时,可能会存在内存分配。
一般来说,最好避免在ISR代码中使用浮点数,硬件设备处理整数转换为浮点数通常在主循环中完成。不过,也有一些DSP算法需要使用浮点数,在带有硬件浮点的平台上(如 Pyboard),可以使用内联 ARM Thumb 汇编程序来绕过这一限制,因为处理器将浮点数值存储在机器字中,ISR和主程序代码之间可以通过浮点数组共享数值。
使用microropython.schedule
可以调度ISR回调在堆未锁定时排队执行,回调函数可以创建Python对象并使用浮点数,也会保证任何Python对象的更新均在主程序完成,因此回调不会遇到部分更新的对象。
其典型用途是处理传感器硬件。ISR从硬件获取数据,发出中断,然后调度回调函数来处理数据。
调度中的回调应符合下文概述的中断处理程序设计原则,这样做的目的是为了避免在主程序循环中抢先执行的任何代码因I/O活动和修改共享数据而产生的问题。
执行时间需要结合中断发生的频率来考虑。如果在执行前一个回调时发生中断,将排队等待执行另一个回调实例,该实例将在当前实例完成后运行。因此,持续的高频率中断会导致队列无限增长的风险,并最终导致运行时错误(RuntimeError)。
如果要传递给 schedule() 的回调是一个绑定方法,请参考 "创建 Python 对象 "中的说明。
当ISR运行时,它可以抢占 asyncio
调度程序。如果 ISR执行asyncio
操作,调度程序的运行就会中断。无论中断是硬中断还是软中断,如果 ISR 已通过micropython.schedule
将执行传递给另一个函数,这一点也同样适用。特别是在ISR上下文中创建或取消任务是无效的。与asyncio
交互的安全方法是实现一个由asyncio.ThreadSafeFlag
执行同步的例程。下面的片段说明了在响应中断时创建任务的过程:
tsf = asyncio.ThreadSafeFlag()
def isr(_): # 中断处理
tsf.set()
async def foo():
while True:
await tsf.wait()
asyncio.create_task(bar())
在本例中,ISR的执行与 foo()的执行之间会有不同程度的延迟。这是协同调度的固有特性。最大延迟取决于应用程序和平台,但通常以数十毫秒为单位。
注意,实时程序的设计错误会导致很难诊断的故障,一般都是没有规律的或者偶发性故障,因此,正确的设和问题预测至关重要。在设计中断处理程序和主程序时,需要注意以下问题。
前面多次提到,ISR的设计要尽可能简单,而且要可预测并在尽可能短的时间内返回,这一点非常重要,因为ISR运行时,主循环并不运行,所以主循环会在代码的随机位置因中断而暂停执行,这种暂停可能会导致难以诊断的错误,尤其是在暂停时间较长或可变的情况下。要了解ISR运行时间的影响,需要知道中断优先级的基本知识。
中断按照优先级方案组织。ISR代码本身可能会被优先级更高的中断中断,如果这两个中断共享数据,则会产生影响(见下面临界区部分)。如果发生这样的中断,ISR 代码就会出现延迟。如果ISR正在运行时发生了优先级较低的中断,它将被延迟到ISR结束,此时如果延迟过长,优先级较低的中断就可能会失败。慢速ISR的另一个问题是,在其执行过程中会出现第二个相同类型的中断,第二个中断将在第一个中断终止后处理,但是,如果接收中断的速度持续超过ISR的处理能力,结果可能会很糟糕。
因此,应避免或尽量减少循环结构。除中断设备外,通常应避免其他设备的I/O,如:磁盘访问、打印语句和 UART 访问等I/O相对较慢且持续时间不同的I/O。还有一个问题是,文件系统函数不是可重入的,在ISR和主程序中使用文件系统I/O是非常危险的。最重要的是,ISR代码不应等待事件发生,如果代码能保证在可预测的时间内返回,I/O是可以接受的,例如切换引脚或LED。另外,若非必要,尽量不要通过I2C或SPI访问中断设备,如果要用,应计算或测量此类访问所需的时间,并评估其对应用程序的影响。
ISR和主循环之间通常需要共享数据。这可以通过全局变量、类或实例变量来实现,变量通常是整数或布尔类型,或整数或字节数组(预分配的整数数组比列表访问速度更快)。当ISR对多个值进行修改时,需要考虑中断发生时,主程序已经访问了部分值,这可能会导致不一致。
在设计时,要考虑以下问题:ISR将接收到的数据存储在字节数组中,然后将接收到的字节数添加到一个整数中,该整数代表准备处理的总字节数。主程序读取字节数,处理字节,然后清空待处理字节数。这个过程一直持续到主程序读取完字节数后发生中断。ISR 会将添加的数据放入缓冲区,并更新收到的数据,但主程序已经读取了数据,因此只能处理最初收到的数据。新到达的字节会丢失。
有多种方法可以避免这种危险,最简单的就是使用循环缓冲区。如果无法使用具有内在线程安全性的结构,下文将介绍其他方法。
如果一个函数或方法在主程序和ISR之间共享,亦或者在多个ISR之间共享,则可能发生潜在危险。这里的问题是函数本身可能会被中断,并运行该函数的另一个实例。如果出现这种情况,函数必须设计为可重入。
临界区的一个例子是访问一个以上可能受ISR影响的变量。如果中断恰好发生在访问各个变量之间,它们的值就会不一致,这就是所谓竞赛条件的危险实例:ISR和主程序循环竞相改变变量。为了避免不一致,必须采用一种方法来确保 ISR在临界区的持续时间内不改变变量值,实现这一目的的方法之一是在该部分开始前执行pyb.disable_irq()
禁止中断响应,并在结束时执行pyb.enable_irq()
允许中断响应。下面是这种方法的一个示例:
import pyb, micropython, array
micropython.alloc_emergency_exception_buf(100)
class BoundsException(Exception):
pass
ARRAYSIZE = const(20)
index = 0
data = array.array('i', 0 for x in range(ARRAYSIZE))
def callback1(t):
global data, index
for x in range(5):
data[index] = pyb.rng() # 模拟量输入
index += 1
if index >= ARRAYSIZE:
raise BoundsException('Array bounds exceeded')
tim4 = pyb.Timer(4, freq=100, callback=callback1)
for loop in range(1000):
if index > 0:
irq_state = pyb.disable_irq() # 临界区开始
for x in range(index):
print(data[x])
index = 0
pyb.enable_irq(irq_state) # 临界区结束
print('loop {}'.format(loop))
pyb.delay(1)
tim4.callback(None)
临界区可以由一行代码和一个变量组成。如下面的代码片段。
count = 0
def cb(): # 中断回调
count +=1
def main():
# 省略了设置中断回调的代码
while True:
count += 1
本例说明了错误的一个微妙来源。主循环中的行计数 += 1 带有一种特殊的竞赛条件危险,称为读-修改-写。这是造成实时系统错误的典型原因。在主循环中,MicroPython读取count
的值,将其加 1,然后写回。在极少数情况下,中断发生在读取之后和写入之前,中断会修改计数,但当ISR返回时,其变化会被主循环覆盖,在实际系统中,这可能会导致不可预测的罕见故障。
如上所述,如果在主代码中修改了Python内置类型的实例,而该实例又在ISR中被访问,那么就应该小心谨慎。执行修改的代码应被视为关键部分,以确保ISR运行时实例处于有效状态。
如果数据集在不同ISR之间共享,则需要特别注意。这里的危险在于,当优先级较低的中断已经部分更新了共享数据时,优先级较高的中断可能会发生。处理这种情况是一个更高级话题,有时可以使用互斥体来解决。
在临界区的持续时间内禁用中断是通常最简单的方法,但它会禁用所有中断,而不仅仅是可能导致问题的中断。长时间禁用中断通常是不可取的。就定时器中断而言,它会给回调发生的时间带来变化。在设备中断的情况下,可能会导致设备过迟得到服务,从而可能造成数据丢失或设备硬件超限错误。与ISR类似,主代码中的关键部分也应具有较短的、可预测的执行时间。
处理临界区段的一种方法是使用一种称为互斥(mutex)的对象(名称来源于互斥概念),这种方法可以从根本上减少中断被禁用的时间。主程序在运行关键部分前锁定互斥器,并在结束时将其解锁。ISR会测试互斥体是否被锁定,如果是,它就会避开关键部分并返回。这个设计的难点在于定义ISR在拒绝访问临界变量时应该做什么,这里有一个简单的互斥体示例。请注意,互斥代码确实禁用了中断,但只禁用了8条机器指令:这种方法的好处是其他中断几乎不受影响。
中断处理程序(如与定时器相关的处理程序)可以在程序终止后继续运行。这可能会产生意想不到的结果,在这种情况下,可能会认为引发回调的对象已经超出了作用域。例如在 Pyboard 上:
def bar():
foo = pyb.Timer(2, freq=4, callback=lambda t: print('.', end=''))
bar()
计时器在被禁用或复位(Ctrl+D)之前,将持续运行。