MicroPython与DMA链式触发结合:ESP32-S3多路串口数据聚合方案
2026/9/7 12:50:03 网站建设 项目流程

最近在做一个多传感器边缘采集盒子,主控用的是 ESP32-S3,程序框架选了 MicroPython,结果遇到一个很现实的矛盾:Python 写业务逻辑是真的快,但数据吞吐和丢包问题也是真的让人头疼。几路 UART 同时上报不定长数据帧,波特率稍高一点,MicroPython 层的中断和轮询就开始轮流虐我,甚至一度想换回 C 从头写。后来换了个思路,把 DMA 链式触发和 Scatter-Gather 这套硬件机制引进来,让 MicroPython 只做配置和数据消费,搬运的工作全部交给硬件,问题一下解决了。

这篇就拆一下这个“MicroPython + DMA 链式触发 + Scatter-Gather 数据聚合”方案,从原理、硬件准备、固件扩展,到 Python 端的聚合逻辑和排查经验,一次性讲清楚。项目本身不是什么高端科研课题,但思路和踩坑过程对做边缘采集、串口多路数据接收这类活儿的同学应该有参考价值。

1. 为什么用 MicroPython 还要折腾 DMA 链式触发

1.1 场景还是那个场景:多路串口不定长数据

先交代一下背景。我这个采集盒子要接六路串口传感器,每一路的数据帧长度不固定,短的十几个字节,长的几百字节,而且各路传感器上报节奏也不一样,有的 10Hz,有的 50Hz。以前用纯 C 写 STM32 的时候,处理这种多路不定长串口数据最常用的套路就是“串口空闲中断 + DMA 接收”,数据搬移几乎不占 CPU,中断只在总线空闲的时候触发一次,效率很高。

但换到 MicroPython 上,情况就尴尬了。MicroPython 本身是解释执行的,Python 层跑一个字节一个字节地读 UART,哪怕开了缓冲区,在高波特率、多路并发的情况下也容易丢数据。用machine.UARTirq写回调,中断频率稍微高一点就顶不住。其实不是 ESP32-S3 性能不行,而是解释器本身承担不了那么高频的中断处理和字节搬运。

所以说到底,问题不是“MicroPython 够不够快”,而是“我用 MicroPython 的方式不对”。数据搬运这种高频率、低智商的体力活,压根不应该让 Python 来做,应该交给 DMA 控制器。

1.2 MicroPython 写不了 DMA,但可以“编”一个 DMA 引擎

MicroPython 的 Python 层确实不能直接操作 DMA 链表寄存器,这一点和 C 不一样。但换个角度想,MicroPython 最擅长的就是“调用别人写好的底层能力”,我完全可以把 DMA 链式触发、Scatter-Gather 这些操作封装成一个自定义的 C 扩展模块,然后在 Python 层调用。

这就是我最终采用的三层架构:

  • 底层:ESP32-S3 的 GDMA 控制器,负责实际的字节搬运和链路切换
  • 中间层:一个由 C 编写的 MicroPython 扩展模块dma_engine,负责配置 DMA 通道、构建描述符链表、触发传输、上报完成状态给 Python
  • 上层:MicroPython 脚本,负责解析数据帧、聚合各路数据、写入 SD 卡或者上报

核心思想就一句话:让硬件做硬件擅长的事,让 Python 做 Python 擅长的事。DMA 链式触发和 Scatter-Gather 把“多段不连续内存之间的高效搬运”这个脏活累活干了,MicroPython 只负责“什么时候开始、数据怎么拼、拼完怎么办”。

1.3 这套设计的适用场景

如果你也在做类似下面这几类项目,这个方案会比较对口:

  • 边缘数据采集盒子,需要接收多路串口、SPI、I2S 上报的传感器数据
  • 需要将多段不连续内存中的数据按帧聚合、重排后统一处理
  • 希望继续用 MicroPython 写业务逻辑,同时不想牺牲底层数据吞吐
  • 做协议转换、数据记录仪、低成本多通道监测设备

反过来说,如果数据量极小、波特率很低,或者业务逻辑对时序不敏感,那就没必要上这套方案,直接用 MicroPython 的 UART 轮询就行,别给自己找麻烦。

2. Scatter-Gather 与链式触发到底解决了什么问题

2.1 从一次搬运说起:DMA 的“外设到内存”

DMA 这件事,玩过 STM32 的哥们都熟,简单说就是外设(UART、ADC、SPI)的数据不经过 CPU,直接由 DMA 控制器搬到你指定的内存地址。CPU 只需要在传输完成之后来处理数据,搬运过程中完全可以去干别的事。

在 ESP32-S3 上,这个控制器叫 GDMA(General DMA)。它比传统单片机上的 DMA 要灵活得多,其中一个很关键的能力就是链表模式。所谓链表模式,就是 DMA 不再只搬运“一段连续内存”,而是可以按照一个描述符链表,依次搬运好几段不连续的内存区域。每搬运完一段,自动跳到下一个描述符继续,全程不需要 CPU 介入。

拿我这次的多路串口接收举例,系统给每一路传感器预分配了多个接收缓冲区,这些缓冲区在内存里是分散的。如果没有链表模式,一个缓冲区满了就得停止 DMA,CPU 介入来换缓冲区,中间这段切换时间就可能丢数据。而在链表模式下,第一个缓冲区满了之后,DMA 会自动走下一个缓冲区,衔接是硬件级别的,几乎没有空隙。

2.2 Scatter-Gather:把散装货物打包成一条运输链

Scatter-Gather,中文常译为“分散/聚集”,它的核心价值在于把分散的内存块组织成一起处理的传输序列。打个比方,你搬家的时候,零碎物件放在好几个房间,DMA 链式触发就相当于一个调度系统,让搬家工人按照预先规划好的路线,依次到每个房间把东西搬上车,全程不需要你一个个房间去催,搬完最后一间它自己就停了。

在多路串口采集里,Scatter-Gather 还能做到“链路绕行”。比如有三路传感器,我可以把三路的接收缓冲区首尾相接,构建一条 DMA 链表,让 DMA 依次接收三路数据,形成一份连续的聚合数据块。这样 MicroPython 层拿到的就不是三个零散的 bytes 对象,而是一整段按顺序拼接好的数据,数据帧重组的工作量会小很多。

2.3 链式触发:描述符链表自动“接单”

DMA 描述符链表的每一跳,都是一个描述符节点。在 ESP32-S3 的 GDMA 驱动里,每个描述符节点保存着三类关键信息:缓冲区地址、缓冲区长度,以及下一跳描述符的地址。链式触发就是让 DMA 控制器顺着这个链表走,走到最后一个描述符之后再触发完成中断。

描述符节点的基本格式大致如下,具体以芯片参考手册为准:

字段作用
buffer address当前这一段搬运的目标内存起始地址
buffer size当前缓冲区容量,即这一段最多能容纳多少数据
next descriptor pointer下一跳描述符的内存地址,形成链表
owner/configuration权限位、突发传输配置、中断使能等控制信息

这个结构非常像操作系统的内存描述符,也很像网卡收发包使用的 DMA ring,只不过 GDMA 把这种能力开放到了通用外设。有了这个机制,MicroPython 层只需要通过 C 扩展模块一次性地把链表建好,之后所有缓冲区切换、数据搬运、链路收尾动作都自动完成,CPU 占用几乎可以忽略。

3. 硬件与底层准备

3.1 选型说明:为什么选 ESP32-S3

做这个项目之前,我对比了几款常用的支持 MicroPython 的芯片平台:

平台优点缺点
ESP32-S3双核 240MHz、GDMA 支持链表、内存充足、Wi-Fi/BLE 可用MicroPython 官方未直接暴露 GDMA API,需要扩展模块
RP2040 (Pico)MicroPython 原生提供rp2.DMA,上手快单核、DMA 复杂度和灵活度相对有限
STM32F4/F7经典 DMA 能力强,资料多MicroPython 端口对 DMA 的 Python 层 API 支持较弱,常用于 C 开发

最终选了 ESP32-S3,主要原因有两个:一是内存和算力都够,MicroPython 跑业务逻辑、跑轻量协议解析、跑简单加密都有余量;二是它的 GDMA 描述符链表模式和 Scatter-Gather 的匹配度很高,底层实现灵活,后续想扩展到 ADC 连续采样、I2S 音频流也顺手。

3.2 定制固件与 C 扩展模块

既然官方 MicroPython 没有把 GDMA 直接暴露给 Python 层,那就要自己动手写一个扩展模块。MicroPython 的 C 扩展模块结构和 CPython 的扩展模块写法非常相似,核心是定义一个 Python 可导入的模块,把 C 函数注册成模块方法。

扩展模块的骨架大致长这样:

#include "py/runtime.h" #include "py/obj.h" #include "esp_private/gdma.h" #include "driver/uart.h" // DMA 引擎对象 typedef struct { mp_obj_base_t base; gdma_channel_handle_t dma_chan; // 其他配置字段 } dma_engine_obj_t; // 构建描述符链表 static mp_obj_t dma_engine_build_chain(mp_obj_t self_in, mp_obj_t buf_list) { dma_engine_obj_t *self = MP_OBJ_TO_PTR(self_in); // 解析 Python 传入的 buffer 列表 // 申请并初始化 GDMA 描述符节点,将这些 buffer 串联成链表 // 返回 chain_id 给 Python 层 return mp_obj_new_int(chain_id); } // 触发链式传输 static mp_obj_t dma_engine_trigger(mp_obj_t self_in, mp_obj_t chain_id) { // 调用 ESP-IDF 驱动接口 // gdma_start(self->dma_chan, (intptr_t)descriptor); return mp_const_none; } // 查询/等待传输完成 static mp_obj_t dma_engine_wait(mp_obj_t self_in, mp_obj_t chain_id) { // gdma_get_interrupt_status / 等待事件组 return mp_obj_new_bool(done); } // 模块方法表 STATIC const mp_rom_map_elem_t dma_engine_locals_dict_table[] = { { MP_ROM_QSTR(MP_QSTR_build_chain), MP_ROM_PTR(dma_engine_build_chain) }, { MP_ROM_QSTR(MP_QSTR_trigger), MP_ROM_PTR(dma_engine_trigger) }, { MP_ROM_QSTR(MP_QSTR_wait), MP_ROM_PTR(dma_engine_wait) }, };

这段代码我只写了骨架,真正实现时还需要处理 buffer 的解析、内存对齐、描述符内存池管理等细节。编译的时候,在 MicroPython 源码的ports/esp32目录下把扩展模块加进CMakeLists.txt,然后重新编译固件,烧录后 Python 端就能直接import dma_engine了。

提示:如果不想自己维护整棵 MicroPython 源码树,也可以把扩展模块放到用户分区,用mpremote配合 manifest 方式打包进固件,这样升级业务代码时不用反复重编固件。

4. 核心实现:MicroPython 数据聚合代码

4.1 DMA 引擎接口设计

C 扩展模块写好之后,Python 层的接口设计就非常关键。好的接口应该让上层调用者不必关心 DMA 描述符、链表这些底层概念,而是用“类对象 + 方法”的方式操作。我最后用的接口是这样的:

from dma_engine import DMAEngine # 初始化 DMA 引擎,配置工作模式和中断回调 engine = DMAEngine(channel=0, mode="rx", on_done=handle_data) # 为某个外设(如 UART1)构建一条 Scatter-Gather 链 chain_id = engine.build_chain( peripheral="uart1", buffers=[buf0, buf1, buf2, buf3], buf_len=256 ) # 启动链式传输,DMA 会依次往 buffers 里填数据 engine.trigger(chain_id) # 等待本轮链路传输完成,返回聚合后的数据视图 data = engine.wait(chain_id, timeout_ms=100)

接口设计思路上,我刻意屏蔽了“描述符”“链表节点”这些字眼,让 Python 层只用chain_id来引用一条链。底层描述符怎么排、多少个节点、节点怎么指向,完全由 C 模块内部管理。这对团队里只写 Python 的同事非常友好,他们只管“哪几个缓冲区是一组”“什么时候开抓”“结果去哪了”。

4.2 Python 端数据聚合逻辑

DMA 硬件把一段段数据填进不同的缓冲区之后,MicroPython 层的核心工作就变成了“把这些分段的数据按正确的顺序拼成一个完整的逻辑帧”。这个过程我管它叫“数据聚合”。

我这边实现聚合时,没有把所有字节都拷到一块新内存再处理,而是采用分段视图的方式:先用memoryview包装每个缓冲区,然后按链路的顺序依次解析。这样既节省了一次大拷贝,也保留了每段数据的边界信息,方便按帧解析。

# 简单的数据聚合器 class FrameAggregator: def __init__(self, chain_id, segments): self.chain_id = chain_id self.segments = segments # DMA 填充后的 buffer 列表 self.cursor = 0 def read(self, n_bytes): """跨缓冲区连续读取 n 个字节,返回 bytes 对象""" out = bytearray(n_bytes) got = 0 while got < n_bytes: seg = memoryview(self.segments[self.cursor]) # 计算当前段还剩多少数据 remaining = len(seg) need = n_bytes - got take = need if need < remaining else remaining out[got:got + take] = seg[:take] got += take self.cursor += 1 return bytes(out) def find_frame(self, start_byte=0xAA, end_byte=0x55): """在聚合后的数据里找一帧完整数据""" # 这里实现帧头帧尾搜索、长度校验 pass

实际项目中,这个FrameAggregator还承担了解析帧头、校验 CRC、重组多路传感数据的任务。因为 DMA 只管“搬”,不知道数据是什么格式,所以上层的帧解析逻辑依然要Python 来做。不过这时候 CPU 压力已经不大了,因为搬移过程零拷贝,Python 层只需要做逻辑判断和字节切片。

4.3 启动与回收流程

DMA 链式触发一个容易踩的坑是:链表构建一次只能跑一轮,跑完之后描述符就回到空闲状态,如果不重新配置,第二轮传输不会自动开始。所以整个流程必须是“配置 - 触发 - 等待 - 回收 - 再配置”的循环。

def acquisition_loop(engine, chain_id, buffers, num_rounds): for round_idx in range(num_rounds): # 重新把 buffers 挂到链上(C 模块内部重置描述符 owner 位) engine.reset_chain(chain_id) # 触发 DMA 链路搬运 engine.trigger(chain_id) # 等待完成,timeout 需要结合实际链路耗时设置 ok = engine.wait(chain_id, timeout_ms=50) if not ok: print(f"round {round_idx} timeout") continue # 聚合数据并处理 aggregator = FrameAggregator(chain_id, buffers) frame = aggregator.find_frame() if frame: handle_frame(frame)

这里的reset_chain是 C 模块里新增的一个方法,作用是把每个描述符节点的 owner 位重新还给 DMA 控制器,同时把缓冲区长度字段重置为初始值。不少第一次接触 GDMA 链表的人会忘记这一步,导致第二次触发时 DMA 认为缓冲区已经写满或者没有权限,数据传输直接失败。

4.4 内存复用与动态链构建

还有一个工程化细节值得说:MicroPython 端的缓冲区不要每次采集都新建,否则容易触发垃圾回收,导致时序抖动。最好的做法是启动时一次性申请一个较大的bytearray池,然后按段切分给各路传感器循环复用。

# 启动时创建 buffer 池 POOL_SIZE = 8 * 256 BUF_SEGMENTS = 8 BUF_LEN = POOL_SIZE // BUF_SEGMENTS buf_pool = bytearray(POOL_SIZE) buffers = [ memoryview(buf_pool)[i * BUF_LEN:(i + 1) * BUF_LEN] for i in range(BUF_SEGMENTS) ]

注意这里传进build_chain的 buffer 必须是整块bytearray切片,并且地址要满足 DMA 的对齐要求。ESP32-S3 的 GDMA 一般要求缓冲区地址按 4 字节对齐,保险起见我建议按 16 字节对齐分配。通过内存池复用,MicroPython 的 GC 压力大幅降低,采集循环的实时性也稳定很多。

5. 常见问题与排查手段

5.1 内存地址对齐与 cache 问题

DMA 的一大堆疑难杂症里,内存对齐和 cache 一致性是最常见的两个。ESP32-S3 虽然是片上 SRAM,但在某些场景下 DMA 和 CPU 同时对同一块缓冲区操作,必须考虑 cache 的影响。

实操时我碰到的现象是:DMA 明明往缓冲区写了数据,但 Python 层读取时发现还是旧值。排查下来是 C 模块里没有在 DMA 完成之后主动做 cache 失效操作。解决办法是,在wait()方法里,完成中断触发之后调用esp_cache_msync()或者执行gma_void对应的 cache 同步接口,确保 Python 层读到的是 DMA 写好的最新数据。

注意:缓冲区优先使用内部 SRAM 的 DMA 能力区域,不要放到 PSRAM 上。PSRAM 访问延迟高且 DMA 支持受限,实测在 PSRAM 上构建描述符链表容易触发总线错误。

5.2 回调在 MicroPython 里不能直接调用?

我在设计 C 模块时,最开始想把 DMA 完成中断直接回调到 Python 函数。结果发现 MicroPython 的中断回调机制有严格限制:在硬件中断上下文里不能执行 Python 代码,否则会直接FATAL ERROR重启。

解决思路有两种:

  • 方案 A:C 模块里只置位一个事件标志,wait()方法阻塞等待这个标志,Python 层按自己的节奏轮询
  • 方案 B:把标志通过micropython.schedule()注册为异步回调,让 Python 代码在 VM 安全点执行
from micropython import schedule from dma_engine import DMAEngine def _on_dma_done(chain_id): print(f"chain {chain_id} done") engine = DMAEngine(channel=0, mode="rx") engine.attach_done_handler(lambda cid: schedule(_on_dma_done, cid))

方案 B 比方案 A 更实时,但要注意回调不能做耗时操作,否则会阻塞 VM。我是两个方案都保留了,简单场景用方案 A,需要实时报告的场景用方案 B。

5.3 链式触发中断丢失

链式传输中途丢中断,也是一个让人怀疑人生的坑。表现是:数据明明已经填满缓冲区,但 Python 层一直等不到完成状态。排查下来有两个原因:

  • 中断标志没有正确清除,导致后续完成中断无法再次触发
  • 描述符里owner位没有按要求置位,DMA 控制器认为当前节点不是自己的,提前停止了链路

解决方案很朴素但有效:在 C 模块处理中断的函数里,先清中断标志,再遍历一遍描述符链表,确认所有节点的owner位状态,打印初始化和重置后的描述符内容用于定位问题。调试好之后,把描述符初始化逻辑固化下来,基本不会再出问题。

5.4 实测性能与优化

最后放一组我本地的实测数据,供大家做个参照。场景是三路 UART 同时接收不定长数据帧,单帧长度 32~512 字节,波特率 460800,连续采集 10 分钟:

方案CPU 占用(估算)丢帧率
MicroPython 纯轮询/中断读串口70% 以上偶尔丢帧,高负载时严重
MicroPython + UART 自带缓冲区轮询30% 左右低波特率稳定,高波特率仍丢
MicroPython + DMA 链式 Scatter-Gather5%~10%10 分钟 0 丢帧

这个数据只针对我当前的硬件和负载,不同配置下会有差异,但总体趋势很明显:把搬运任务交给 DMA 之后,CPU 占用和稳定性完全不是一个级别的。如果后续还要扩展更多路串口,这套架构的扩展空间也足够,只需要把新的 UART 接到 GDMA 通道上,然后 Python 层加一个chain_id的映射即可。

写在最后的一点体会

这个项目做下来,我最深的感受是:MicroPython 不等于低性能,关键看你把 Python 用在哪个层次。数据搬运这种高频、简单、机械的工作,交给硬件 DMA 是最优解;数据解析、协议处理、业务逻辑这种复杂、低频、需要灵活性的工作,交给 Python 又舒服又高效。两者结合,就形成了“底层 DMA 链式触发 + 上层 Python 数据聚合”的分工模式。

如果你也在做多路串口采集或者其他需要高吞吐数据聚合的 MicroPython 项目,建议先别急着换 C 或换 RTOS,可以试着看看自家平台的 DMA 能不能通过扩展模块暴露给 Python。这个方向的投入回报率其实挺高的,一次架构打通,后续加传感器、加协议都只是在 Python 层做增量,开发速度能快不少。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询