ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

MicroPython中使用DMA链式触发与Scatter-Gather实现高效串口数据接收

MicroPython中使用DMA链式触发与Scatter-Gather实现高效串口数据接收 在MicroPython里提DMA很多人的第一反应是“这货不是给C语言准备的吗”。我一开始也这么想直到一个需要接高频串口数据的项目里被UART.read()的卡顿整到怀疑人生。数据一来主循环就被拖住稍微一忙就丢帧CPU一会儿忙着搬数据一会儿又忙着解析根本没有余力去管别的外设。后来我把ESP32的GDMA链式描述符翻出来在MicroPython里用machine.mem32去摸寄存器再配合Scatter-Gather的思路把一堆小的bytearray串成一条逻辑上的连续流一次DMA搬运完成一整轮数据聚合CPU从搬运工变成了验收员。这篇就是这套方案从踩坑到跑通的完整记录也顺便把DMA链式触发、Scatter-Gather聚合这些听起来吓人的概念用能给小白讲明白的方式拆开。1. 设计思路为什么要折腾DMA链式触发1.1 串口高频数据把主循环打爆之后先说我遇到的场景。我在做一个带OLED显示的小采集器ESP32-S3负责从串口接收一台上位机不断发来的传感器包。包不长20到200字节不等但频率高一秒钟能来几十上百包。用MicroPython最简单的做法是写一个while循环里面不断调用uart.read()或uart.readline()有数据就处理。问题很快就来了。只要数据量稍微大一点主循环就会被读数据、拼包、解析这一连串操作占满OLED刷新、按键扫描、日志输出全部跟着卡顿。我开始以为是MicroPython效率低后来用逻辑分析仪看了一下发现问题核心不在解释器而在于每次来数据都触发CPU中断去搬数据。MicroPython的UART.read()底层走的是中断FIFO数据到了一点就通知CPUCPU就要从当前任务里切出来把小批量数据搬到用户buffer。当数据频繁到达时CPU大多数时间都花在“搬快递”上真正该干的活反而干不了。这时候就想到DMA。DMA的全称是Direct Memory Access它可以在不占用CPU的情况下让外设和内存之间直接传数据。但MicroPython默认并没有把DMA接口暴露出来。不过没有接口不代表不能用ESP32系列的寄存器是可以在MicroPython里用machine.mem32直接读写的。换句话说只要你知道该往哪个寄存器写什么值照样能把DMA拉起来干活。1.2 Scatter-Gather的本质把碎片拼成逻辑流Scatter-Gather中文常叫“分散/聚集”名字听着很玄其实思路特别朴素。比如你需要接收一块1KB的数据。在C语言里你可以直接malloc(1024)拿一大块连续内存。但在MicroPython这样的脚本环境里内存管理是GC堆负责的你没法保证每次都能申请到一大块连续内存尤其是跑了一段时间后堆里到处都是碎片。一次分配1024字节不难但要长期稳定地、反复地分配几块大buffer碎片问题就会冒出来。Scatter-Gather的思路是我不要求一块连续内存我准备4个256字节的小buffer让DMA先填第一块填满了自动去填第二块再填第三块、第四块。四块加在一起是1024字节虽然在物理内存上它们不连续但对于处理数据的逻辑来说它们是按顺序排列的一条连续流。这就叫“聚合”。这样做的好处非常明显。第一内存压力小小块buffer想怎么分配都行GC容易满足。第二接收过程可以持续不断DMA每填满一块就直接去下一块不需要CPU参与。第三所有数据都在内存里之后CPU只需要在DMA完成一轮之后按照每块记录的实际长度把数据取出来做统一解析。1.3 三种方案对比与选型在开始动手之前我整理过三种做法最后选了DMA链式这里把对比放出来给大家做个参考。方案优点缺点适合场景纯轮询read()代码最简单几行就能跑CPU占用高数据一多就丢帧或者卡死低速、低数据量、对实时性要求极低串口空闲中断缓冲区保留MicroPython的易用性支持不定长分包仍然靠CPU搬数据数据量大时CPU占用依然高中低速数据需要帧边界判断DMA链式Scatter-Gather搬运过程零CPU参与支持多块聚合吞吐量大需要操作寄存器代码复杂调试有门槛高频数据、多块缓冲、需要释放CPU的场景我最终选第三种不单是因为吞吐量更因为它能把“搬数据”这件事从CPU手里彻底拿掉。做完之后主循环里再也不用频繁响应串口中断处理逻辑可以一帧一帧地批量跑整个系统的稳定性上了一个台阶。2. 底层原理链式描述符是这样干活的2.1 描述符就是给DMA用的“快递面单”想用好DMA第一个绕不开的概念就是描述符。描述符英文叫descriptor你可以把它理解成一张快递面单。面单上写了“这一单要从哪里取货”、“送到哪里去”、“这一单有多少货”以及“送完之后下一单在哪”。DMA控制器不直接知道你的buffer长什么样它只认描述符。每张描述符通常对应一块内存buffer。你把buffer的地址、长度写进描述符再把描述符的地址告诉DMA控制器。DMA启动后就照着当前描述符的地址去搬运数据。等这一块搬完它再读取描述符里的“下一张描述符地址”继续干下一单活。这个过程自动发生不需要CPU介入就是所谓的“链式触发”。2.2 描述符结构逐位拆解不同芯片的描述符结构略有差异但核心字段大同小异。我以ESP32经典DMA描述符为例它是16字节也就是4个32位字。这里逐字段拆一下。字段位范围说明sizebit0-11当前描述符对应buffer的最大容量单位字节lengthbit12-23DMA实际搬运的数据长度传输完成后由硬件更新offsetbit24-28在buffer内的偏移一般填0sosfbit29start of stream流开始标志在需要标记起始时置1eofbit30end of frame最后一个描述符置1表示整轮传输结束ownerbit31归属位1表示DMA拥有该描述符0表示CPU可以处理后面三个word依次是数据缓冲区地址buf_ptr、下一个描述符地址next、保留字段。用C结构体看会更直观typedef struct { uint32_t size: 12; uint32_t length: 12; uint32_t offset: 5; uint32_t sosf: 1; uint32_t eof: 1; uint32_t owner: 1; uint32_t buf_ptr; uint32_t next; uint32_t reserved; } dma_desc_t;注意不同型号芯片的保留位定义不一定一样。我按ESP32经典布局来写ESP32-S3的GDMA描述符字段基本一致但个别位可能有差异。真正动手前一定打开对应芯片的Technical Reference Manual把描述符布局核对一遍。这里最关键是owner位。DMA搬运数据的时候需要先确认owner为1它才有权限操作这个描述符。搬完之后owner会被硬件清零表示“这个描述符的数据我已经弄完了CPU你可以来处理了”。CPU处理完数据之后需要把owner重新置回1表示“下一轮传输可以使用我了”。这个机制保证了DMA和CPU不会同时操作同一块buffer。2.3 链式触发与环形链表链式触发说起来也简单。你把多个描述符用next字段串起来第一个的next指向第二个第二个的next指向第三个最后一个的next可以指向第一个形成环形。形成环形有一个额外的好处当DMA完成一轮传输之后如果你重新把每个描述符的owner都置1再触发一次启动DMA又能从链头开始继续跑。这样只要软件处理得够快就能实现连续不断的流式接收。很多连续采集场景就是这么干的DMA在幕后一圈一圈地循环CPU只负责在合适的时机来取数据。实际操作中我通常会让最后一个描述符的eof置1。这样当DMA走到最后一个描述符并完成传输后硬件就会产生一个对应的结束信号我可以通过轮询状态寄存器或者中断知道“这一轮数据已经完整落到内存里了”。如果是环形链下一轮又从第一个描述符开始数据会覆盖上一轮的内容所以解析数据必须赶在下一轮覆盖之前完成。2.4 为什么在MicroPython下更要注意细节MicroPython给人的印象是“轻松写业务逻辑”但摸寄存器这事它跟C语言一样“裸”。machine.mem32读到多少就是多少写错寄存器不会像Python那样抛异常而是直接让芯片进入未知状态最常见的就是死机或总线错误。在MicroPython里做DMA最大的风险还不是寄存器写错而是GC捣乱。MicroPython的GC堆会回收不再使用的对象。如果一个bytearray没有变量引用它GC就会把它的内存回收掉之后DMA还在往那个地址写数据就会踩到已经被回收甚至被重新分配的内存数据损坏、死机都算轻的。更隐蔽的是有些MicroPython固件支持PSRAM某些对象会被分配到外部RAMDMA访问外部RAM的延迟和一致性问题又会让调试雪上加霜。所以在这套方案里内存管理和引用保持比寄存器配置更值得花精力。3. MicroPython实现从描述符到寄存器3.1 先解决两个前置问题地址与引用在MicroPython里获得一块bytearray的底层内存地址有个很方便的办法用uctypes.addressof()。它返回对象的缓冲起始地址这个地址可以直接写进描述符里。from uctypes import addressof buf bytearray(256) addr addressof(buf) print(hex(addr))这里有几个细节要记住。第一addressof()拿到的地址在对象存活期间是稳定的。MicroPython的GC不会像某些运行时那样移动已分配对象所以只要你还持有这个bytearray的引用地址就一直有效。第二必须用变量持续引用住这些bytearray否则它们会被GC回收。我一开始就是吃了这个亏描述符建好了DMA跑起来了但接收缓冲区已经被GC回收了数据全写到别的地方去了。后来我把所有buffer和描述符统一放到一个类实例里让类生命周期管理它们的引用问题就解决了。3.2 描述符构造与链表绑定构造描述符不复杂核心是按位填第一个word。下面是我在项目中用到的构造逻辑。import struct from uctypes import addressof # 位定义按经典lldesc布局 OFF_SIZE 0 OFF_LEN 12 OFF_SOSF 29 OFF_EOF 30 OFF_OWNER 31 def make_descriptor(buf_addr, buf_size, eof0, owner0, next_addr0): word0 (buf_size 0xFFF) | ((buf_size 0xFFF) OFF_LEN) if eof: word0 | (1 OFF_EOF) if owner: word0 | (1 OFF_OWNER) desc bytearray(16) struct.pack_into(IIII, desc, 0, word0, buf_addr, next_addr, 0) return desc构造时我把length初始成和size一样是一个常见的做法。DMA开始搬运后硬件会根据实际搬运量把length改成真实字节数。数据解析时我们读取的就是这个字段。然后是绑定链表。把所有描述符的next指针串起来最后一个视情况让它指向第一个做环形结构。def bind_chain(descs): n len(descs) for i in range(n): next_addr addressof(descs[(i 1) % n]) struct.pack_into(I, descs[i], 8, next_addr) if i n - 1: # 最后一个作为本轮结束标志 w0 struct.unpack_from(I, descs[i], 0)[0] w0 | (1 OFF_EOF) struct.pack_into(I, descs[i], 0, w0)注意struct.pack_into(I, ..., 8, ...)写入的是描述符的第三个word也就是next字段。3.3 DMA通道初始化与启动到了这一步不同芯片的寄存器细节差异就大了。我不把这里写成一份“照抄就能跑”的代码因为寄存器偏移量必须和你手上的芯片、SDK版本、开发板对应。我把自己梳理的初始化步骤写出来你照着步骤去查自己芯片的手册比自己瞎试要快得多。初始化DMA输入通道也就是外设往内存搬运的通道通常需要做以下几件事外设侧先安排好比如UART的RX引脚、波特率、FIFO触发阈值。把DMA通道的链路地址寄存器设为链首描述符的地址同时置链路使能位。配置DMA通道的工作模式让外设作为数据源内存作为目的地。如果有中断需求使能完成中断或者错误中断。在MicroPython里处理中断没有C那么方便所以我用轮询方式更多。在MicroPython里写寄存器就是一句话的事import machine machine.mem32[DMA_LINK_ADDR_REG] chain_head_addr | 1 # 低位置1表示使能链路但要命的是DMA_LINK_ADDR_REG是多少。不同芯片甚至同一芯片的不同通道都可能不一样。我建议你打开ESP-IDF里的头文件比如esp32s3/include/soc/gdma_reg.h直接搜in_link或者link_addr把宏定义里的偏移值抄出来。千万别在网上随手抄一段地址就开跑。配置好之后启动DMA实际还不够。如果你用的是UART这类外设还需要让外设的DMA请求使能。比如UART的RX要设置成DMA模式数据到了FIFO之后自动触发DMA搬运而不是触发CPU中断。这一步一般是往外设的控制寄存器里写DMA使能位和FIFO阈值。3.4 等待完成与数据聚合启动之后传输是DMA自己跑的。我们需要检测“一轮是否结束”。最直接的办法是轮询最后一个描述符的owner位。当owner被硬件清零说明该描述符已经被DMA用过数据已经到位。def wait_done(descs, timeout_ms1000): last descs[-1] start time.ticks_ms() w0_offset 0 while time.ticks_diff(time.ticks_ms(), start) timeout_ms: w0 struct.unpack_from(I, last, w0_offset)[0] if (w0 (1 OFF_OWNER)) 0: return True return False等到一轮结束就可以遍历所有描述符读取它们的length字段然后把对应buffer里的有效数据拼起来或者直接逐个解析。def collect(blocks, descs): for index, desc in enumerate(descs): w0 struct.unpack_from(I, desc, 0)[0] length (w0 OFF_LEN) 0xFFF if length 0: continue block blocks[index] # 这里可以根据具体帧格式解析 block[:length] process_frame(bytes(block[:length]))解析完之后要把owner全部置回1准备下一轮传输def reset_owner(descs): for desc in descs: w0 struct.unpack_from(I, desc, 0)[0] w0 | (1 OFF_OWNER) struct.pack_into(I, desc, 0, w0)然后重新启动链路继续下一轮。这里有个经验如果数据是连续的最好在解析完成之前就把owner重新置1让DMA可以尽早开始下一轮避免出现间隔期漏数据。3.5 完整代码骨架把上面这些串起来一个相对完整的Scatter-Gather DMA模块骨架如下。寄存器地址部分我用占位符表示真正使用前一定要替换成自己芯片的实际地址。import struct import time import machine from uctypes import addressof # 位定义 OFF_LEN 12 OFF_EOF 30 OFF_OWNER 31 # 平台相关寄存器地址务必按实际芯片/手册填写 DMA_LINK_ADDR_REG 0x00000000 # 替换为你的链路地址寄存器 DMA_CH_CONF_REG 0x00000000 # 替换为你的通道配置寄存器 UART_DMA_CONF_REG 0x00000000 # 替换为你的外设DMA使能寄存器 class ScatterGatherDMA: def __init__(self, block_count, block_size): self.block_count block_count self.block_size block_size self.blocks [bytearray(block_size) for _ in range(block_count)] self.descs [bytearray(16) for _ in range(block_count)] self._build_chain() def _build_chain(self): for i in range(self.block_count): buf_addr addressof(self.blocks[i]) next_addr addressof(self.descs[(i 1) % self.block_count]) eof 1 if i self.block_count - 1 else 0 w0 (self.block_size 0xFFF) | ((self.block_size 0xFFF) OFF_LEN) if eof: w0 | (1 OFF_EOF) w0 | (1 OFF_OWNER) struct.pack_into(IIII, self.descs[i], 0, w0, buf_addr, next_addr, 0) def start(self): chain_head addressof(self.descs[0]) machine.mem32[DMA_LINK_ADDR_REG] chain_head | 1 def wait_done(self, timeout_ms1000): last self.descs[-1] start time.ticks_ms() while time.ticks_diff(time.ticks_ms(), start) timeout_ms: w0 struct.unpack_from(I, last, 0)[0] if (w0 (1 OFF_OWNER)) 0: return True return False def process(self): for i, desc in enumerate(self.descs): w0 struct.unpack_from(I, desc, 0)[0] length (w0 OFF_LEN) 0xFFF if length 0: self._handle_block(i, length) self._reset_owner() self.start() def _handle_block(self, index, length): # 每个项目的数据解析逻辑不同这里留给你自己写 chunk bytes(self.blocks[index][:length]) # 比如self.parser.feed(chunk) def _reset_owner(self): for desc in self.descs: w0 struct.unpack_from(I, desc, 0)[0] w0 | (1 OFF_OWNER) struct.pack_into(I, desc, 0, w0)这个骨架里process()每次只处理一个轮次的数据。如果你的数据量非常大来不及在下一轮覆盖前处理完有两种解法一是把block数量加多给CPU更多时间窗二是把_handle_block里的数据快速拷贝到另一个更大的处理队列里而不是原地解析。4. 实测串口不定长数据的Scatter-Gather聚合4.1 测试环境与数据场景我把这个方案实际接到了一个项目里板子是ESP32-S3开发板MicroPython固件是当时官方最新版。数据源是一块USB转串口模块由PC端程序模拟上位机发送不定长数据帧。帧格式很简单帧头0xAA 0x55长度字段1字节数据字段若干字节校验1字节累加和一帧的长度从7字节到100字节不等模拟的是传感器包。PC端每5毫秒发一帧相当于持续不断的负载。DMA侧我配了4块缓冲区每块256字节总聚合长度1KB。也就是说DMA可以连续接收1KB的数据而不需要CPU介入。正常情况下1KB可以装下好几帧数据。4.2 操作流程与验证步骤实际操作步骤整理如下先用UART.read()做个最原始的接收确认串口硬件本身没问题波特率、数据格式都对。把UART的RX模式切到DMA模式。这一步很重要MicroPython的UART.read()默认走的是单片机的RX FIFO加中断不会触发DMA。初始化Scatter-GatherDMA对象构建描述符链表。启动DMA链路。在主循环里定时检查wait_done()。如果完成调用process()解析。在验证阶段我特意让PC端连续发了1000帧数据统计MicroPython端成功解析出来的完整帧个数并且用machine.mem32读了DMA状态寄存器确认DMA确实在不断搬运。实际跑下来每一轮数据到达的间隔在毫秒级CPU完全没有因为数据接收产生明显卡顿。4.3 与轮询模式对比性能差多少这组数据是我在同一块板子上、同一个波特率下做的对比。方式CPU参与度主循环卡顿帧丢失情况可扩展性轮询read()持续占用明显数据多时偶发漏帧差串口中断缓冲中高占用有偶尔漏帧一般DMA链式Scatter-Gather几乎为零无连续1000帧无丢失高单看速度MicroPython的read()其实不慢瓶颈在于“频繁打扰CPU”。DMA的好处不是把速度提升十几倍而是把CPU从“每来几个字节就响应一次”的状态里解放出来。做完这个改造之后我的OLED刷新稳定了按键响应也没了卡顿感。4.4 缓冲区数量与大小的选择建议聚合效果和缓冲区配置关系很大。我用不同参数做了几轮测试得出几个经验。配置数据延迟内存占用推荐场景2块 × 256字节低小偶发小数据包4块 × 256字节中中持续中速数据流8块 × 1024字节较高大大块数据、突发流量具体选择时要考虑两个因素一是数据到达的周期二是你处理一帧数据需要的时间。要保证在DMA把一圈描述符都写满之前CPU能完成上一轮数据的解析。否则就会发生覆盖也就是数据还没处理就被下一轮覆盖了。最简单的经验是缓冲区总量至少要能装下3到5倍的“单次最大数据量”这样CPU有充足的时间窗口。5. 常见问题与排查技巧实录5.1 DMA启动后完全不动这几乎是所有人都会遇到的第一道坎。DMA启动后状态寄存器始终是0数据一点都没搬。我的排查顺序是这样的确认DMA链路地址寄存器里写入的链首地址和addressof(descs[0])打印出来的值一致。确认描述符里的buf_ptr和addressof(blocks[i])一致。确认外设的DMA请求已经使能。UART如果不切到DMA模式DMA通道永远不会得到触发。确认owner位是1。如果owner是0DMA会认为描述符不属于自己直接不干活。很多时候问题出在第三步UART还工作在普通FIFO模式数据到了FIFO之后只会触发常规中断DMA根本不会感知到。5.2 数据错位、长度对不上数据错位一般是描述符和buffer对应关系错了。比如描述符1的buf_ptr填成了buffer 2的地址或者next指向了错误的描述符DMA就把数据写到了意料之外的地方。排查办法是启动之前手动把所有描述符的word都打印出来逐个核对地址。尤其是buf_ptr和next最好写一个小循环把每个描述符的字段dump一遍再启动。MicroPython的交互式环境做这件事特别方便别偷懒。长度对不上则大概率是length初始值不是size。有些芯片要求length初始为0有些要求等于size差别就在DMA搬运完成之后length更新成的值是否等于实际搬运字节数。这个问题必须看手册确认。5.3 一轮之后第二次启动失败第一轮数据正常第二轮开始却收不到数据。这个我调试时也遇到过。根因通常是owner位没有全部重置。第一轮结束后所有描述符的owner位都会被硬件清零。你需要在处理完数据之后把所有描述符的owner重新置回1再重新启动链路。如果只重置了链首DMA走到第二个描述符时发现owner不是1就停在那里了。另外如果最后一个描述符的eof位没有清零第二轮启动后可能刚到最后一个描述符就触发了结束中断表现就是第二轮数据不完整。5.4 GC回收导致描述符被改这是MicroPython特有的坑。出现特征是DMA跑着跑着数据开始乱掉甚至系统直接复位。原因是描述符本身或者buffer被GC回收了。我在前面强调过要持有引用这里再补一个细节即使你持有引用也要保证在DMA运行期间没有大量执行会触发GC的操作。MicroPython的GC是自动的分配新对象时就可能触发。如果你在接收循环里不断创建临时对象GC可能随时跑一遍。虽然它不会移动对象但会回收掉所有没有根引用的对象其中就可能包括某个你只在启动时用了一次的临时bytearray。我的习惯是所有描述符和buffer在初始化时都放进类实例的属性里业务处理过程中不再重新分配这些核心对象。解析数据时如果产生了临时对象也尽量在DMA空闲时段处理而不是在数据风暴高峰处理。5.5 问题速查表现象大概率原因解决方法DMA完全不动外设DMA请求没使能检查外设控制寄存器DMA不动且owner为0描述符被CPU处理过但没重置owner重置所有owner再启动数据写到奇怪的地方buf_ptr填错或者buffer被GC回收打印描述符字段核对地址第二轮开始数据异常描述符eof/owner状态没复位统一重置再启动数据长度不对length初始值和芯片要求不符查手册调整初始值系统周期性复位DMA写入地址越界检查buffer大小和描述符size字段是否匹配最后分享一个我自己踩出来的小技巧。这套方案在MicroPython里调试确实比C费劲尤其是寄存器地址不确定的时候别硬猜。先在ESP-IDF的C工程里把DMA跑通再回到MicroPython照抄逻辑对MicroPython代码里每一条寄存器操作都能在C例程里找到对应。这样两边一对照哪儿写错很快就能定位。我后来在几个不同类型的采集项目里都用上了这套方法把串口、ADC、I2S的数据流都换成了DMA链式聚合CPU占用率肉眼可见地降了下来。如果哪次传输的时机要求特别严格我还会考虑用定时器中断去配合DMA的完成事件把精度再往上提一档。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进