企业官网建设流程全解析

QNX开发之ECAT专用网卡驱动ecpkt · 05-接收

上一章结束时驱动的发送路径已经闭环：50 条测试报文全部被 PC 端收到，write() 一路畅通。但驱动还一帧报文都收不到——接收路径还是空的。这一篇我们来实现接收路径，把报文从网线一路送到应用程序的 read() 里。

在发送篇里我们已经理解了 SG-DMA 和 bdring，接收功能同样是基于这两个模块。它与发送功能不一样的地方在于：发送是由驱动主动发起，接收则是被动响应的。这种被动响应的处理机制一般是周期轮询或中断，两种方案各有优点：周期轮询对应用的节奏控制更稳定，但数据处理会有一些延迟；中断模式的数据处理更及时，但中断会打断应用程序的节奏，且因为要走中断处理器，系统开销也更大。这里我们沿用 devnp 的中断方案，在 RX 中断里处理接收事件。

注：实际上对于 EtherCAT 通信来说，发送和接收的节奏都是固定可控的，使用周期轮询机制反倒实时性会更好。比如 Ubuntu 上的主站方案 IGH 也会为某些网卡芯片（比如 Intel 的 e1000 等）提供专用的实时驱动，这些实时驱动相比通用驱动的一大改动就是把中断改成了轮询（这里有一部分原因是 Linux 的中断处理属于软实时而不是硬实时，会引入较大的抖动）。我们这里先使用中断模式快速完成 ecpkt 的开发，后续如果要进一步调优再考虑改成周期轮询的方式来实现接收功能。

1. 同样基于SG-DMA和bdring的接收路径

把发送路径倒过来就是接收路径，SG-DMA、bdring、frame pool 这套机制全部原样复用，只是数据流向反过来。发送是 write() 把用户数据写进槽位，驱动把槽位的物理地址和长度填进 BD，DMA 沿 bdring 把数据搬给 MAC 发出去；接收则是 MAC 收到帧，DMA 沿 bdring 找到 BD 指向的槽位把帧写进去，驱动再把这一帧交给接收队列，最后 read() 从队列里交给应用程序。

接收路径和发送路径的主要区别在于槽位的生命周期：发送的槽位是“过路的”——write() 时借一个，发完就还，槽位在 FREE、READY、INFLIGHT 之间流转；接收的槽位则必须是“钉死的”——DMA 随时可能把一帧写到任何一个 BD 指向的槽位里，如果槽位和 BD 的对应关系像发送那样动态变化，DMA 在激活 BD 时 BD 里还没有可用地址，DMA 就无法完成数据搬运，对应的报文就直接丢了。

所以接收路径下必须要保证可用 BD 里的地址是可用的。我们在初始化时一次性地把所有接收槽位和 BD 绑死——BD j 永远指向 slot j，槽位的物理地址写进 BD 之后就不再改变；BD 的状态切换与槽位的状态绑定在一起，bdring 的更新依赖于槽位状态的更新，“槽位可用”是 BD 可用的前提。为了区分这两种语义，我们给 frame pool 的 slot 加了第五个状态 SLOT_BOUND（绑定态）：发送走借还（alloc / release），接收走绑定（bind / unbind），两套生命周期在同一个池里并存。

另外，RX 环上任何时刻不能有空 BD。DMA 收到帧时是自己找 BD 写数据，找不到可用的 buffer 就直接丢包，不会等谁——所以初始化时必须把 128 个 BD 全部挂满 buffer 交给硬件，同时每处理一帧，回收 BD 和补挂新 buffer 必须成对出现。把 BD 从硬件收回来，就要立刻重新挂回去；少这一步，这个槽位从此就收不到包了。

2. 中断处理

中断挂载与使能我们直接沿用 devnp 里的实现：

iid = InterruptAttach(xzynq->irq, xzynq_isr, arg, 0, _NTO_INTR_FLAGS_PROCESS);

devnp里的中断处理包含了中断上半段和中断下半段的处理：

const struct sigevent *xzynq_isr(void *arg, int iid) { xzynq_dev_t *xzynq = arg; struct _iopkt_inter *ient = &xzynq->inter; /* Disable all interrupts */ out32(xzynq->regbase + XZYNQ_EMACPS_IDR_OFFSET, XZYNQ_EMACPS_IXR_ALL_MASK); /* Clear interrupts */ xzynq->isr_status |= in32(xzynq->regbase + XZYNQ_EMACPS_ISR_OFFSET); out32(xzynq->regbase + XZYNQ_EMACPS_ISR_OFFSET, xzynq->isr_status); return interrupt_queue(xzynq->iopkt, ient); }

中断处理函数xzynq_isr完成上半段的寄存器操作之后，往iopkt的任务队列里添加了一个事件，相当于把下半段交给iopkt来处理，iopkt的任务队列有点类似于linux下的workqueue，属于iopkt特有的机制，在ecpkt里面就需要我们自己来实现，不过iopkt的任务队列机制是针对多网卡多应用的复杂情况来处理的，ecpkt里可以大大简化，我们直接起一个线程来处理中断下半段就可以了：

pthread_create(&g_drv.int_thread_t, NULL, int_thread, &g_drv.xzynq)

补充一下：开发接收功能时，我想当然地把中断处理和接收数据分成两个模块，屏蔽掉接收数据先专注调试中断——比如中断能否触发、中断处理路径是否正常执行等。但 Zynq 的 GEM 控制器上，中断触发并不是在 GEM 收到数据时，而是在 DMA 把数据完成搬运之后——中断触发的语义不是“有数据到达”，而是“有数据可用”。如果没有接收数据逻辑，bdring 里的 BD 不会被正常释放，BD 很快就会耗尽，从而 DMA 无法完成数据搬运，进而无法触发“可用数据”中断。所以中断处理和数据接收两个功能是强耦合在一起的，两者必须协同起来才能工作。

基于上面两点，接收路径要做的事情就清楚了：

• 初始化时把所有接收槽位预挂给硬件（全占满）
• 中断里收帧：排空接收环、记录长度、交给接收队列、成对补齐
• read() 把帧从接收队列交给应用程序

下面我们具体展开这三件事。

1. 预挂接收槽位

初始化阶段对每个空闲 BD 做三件事：从 bdring 申请一个 BD；如果这个 BD 是第一次挂接（rx_slot_idx[j] 还是 NONE），就把 slot j 和它绑定，并把槽位的物理地址写进 BD——这个地址从此不再改动；最后把这个 BD 交给硬件。128 个 BD、128 个槽位、每槽 2048 字节，一轮循环全部挂满。

之后每次收帧后的补挂走的是同一段代码的另一条路径：槽位和 BD 地址都不动，只是把 BD 重新交给硬件。整个接收数据路径上没有任何 alloc / release——绑定态的槽位永远不进空闲栈，也不会被发送路径误用。

2. 中断里收帧

中断上半段里完成现场保护：读 ISR 寄存器并写回清除中断源（GEM 的中断是电平触发，源不清、中断线就一直拉高）、屏蔽中断（防止下半段处理期间 ISR 反复重入空转）、返回一个 sigevent 唤醒下半段。

下半段线程化：InterruptWait 被唤醒后进 process_interrupt，发现是帧接收完成（FRAMERX）就调 process_rx 排空整个接收环——from_hw_rx 把所有已完成的 BD 一并取回；逐帧校验 SOF / EOF；按 rx_slot_idx[j] 找到槽位（发送那张是 BD 正持有哪个槽位的反向映射，接收这张因为绑定恒成立，本质上是一个恒等映射）；把 BD 报告的真实长度记到槽位上，供后面的 read() 查询；最后 rx_push 把这一帧交给接收队列。一轮的收尾是成对补齐：bdring_free 回收这批 BD，setup_rx_buffers 把它们重新挂回硬件；如果这期间又到了新帧，继续下一轮循环。

3. read() 取数据

中断线程把帧塞进接收队列——一个 128 项的环，每项记录槽位号和长度；用户态的 read() 从环的另一头取。队列满时丢帧计数：push 跑在中断线程上下文，绝不能阻塞等待。用户 buffer 比帧小时返回 EMSGSIZE，帧留在队头不消耗，下次 read() 还能取到——“读缓冲太小”不该吃掉一帧数据。

数据本身的交付是零拷贝的：帧还在 NOCACHE 映射的槽位内存里，read() 处理时直接用 _RESMGR_PTR 把槽位内存挂到回复的 iov 上，客户端经内核取走，全程没有一次 memcpy。

这里有一个需要注意的地方：read 操作是阻塞的，等待数据到来时进程会挂在 read 里等事件。这就会带来一个问题——RM 的消息分发是单线程的，等待 read 时 RM 的消息处理 dispatch 是无法被激活的。这种情况下如果一个客户端意外死亡，正常情况下 RM 会收到一条消息交给 dispatch 去处理；但由于此时 RM 阻塞在 read 里，dispatch 不会运行，这个客户端死亡的事件就无法被处理，RM 就会一直阻塞在 read 里直到有数据到来。在此期间所有需要 dispatch 处理的消息（比如新客户端连接的到达）都无法被处理。直观来说，就是用户程序退出后再次启动，请求打开 RM 设备时无法响应。

解决这个问题的办法是延迟回复：把这次请求的 rcvid 停进一个 pending 数组，返回 _RESMGR_NOREPLY——客户端那边看起来还在正常阻塞，RM 的 dispatch 线程却已经回到分发循环继续干活；新帧到达时驱动给自己投一个脉冲，dispatch 线程醒来后由 service_pending 把帧交给停着的那个读请求。对客户端来说，这就是一次普普通通的阻塞 read()。

完成以上工作后，接收功能就可以闭环验证了。我们用下面这套方法测试一下：

• PC 端用小工具发送确定性的广播流量
• 板端运行 gem_test：阻塞 read() 收帧，每收一帧打印摘要

运行日志：

可以看到，PC 发出的报文全都收到了。到这一篇为止，驱动的数据收发通路就完整了，驱动本身的功能开发也告一段落；后续可以使用 SOEM 主站来基于这个驱动进行通信并测量实时性，看看专用驱动的效果怎么样。