【分享帖】使用 FIFO 优化发送程序
使用 FIFO 优化发送程序
Tips: 阅读前请自动 #include <stdint.h>
Tips2: 本文写自某老灯，希望能为 RM 的发展做出一点微薄的贡献，让大家不要再为一些底层的琐事困扰。
Tips3: 本人水平有限，能力不强，仅作为抛砖引玉，提出一种可能的方案，大佬轻喷。
一、背景
使用某种通讯接口进行发送是单片机开发最常见的场景之一。然而通讯速率不同会影响发送策略。
例如当通讯接口速率较快、且发送数据较少时（如USBFS），发送可以看作瞬间完成，反映到代码中：可以认为调用完这次发送代码后，便可以立刻再次调用下一次发送；CPU 下次访问发送寄存器/缓存之前，寄存器/缓存已经清空。也就是连着写几行发送也没有什么大问题。
但如果不幸的是我们在使用某种速率较慢的接口，且一次要发送大量数据时，发送就会遇到大问题。例如使用波特率 115200、8 数据位、1 停止位、无校验位的串口，发送 3 个 float32 类型的数据。帧格式仿照某比赛使用的“串口协议手册”中的格式，我们可以计算出：
- 发送总 Byte 数：5 Byte 帧头 + 2 Byte cmdId + 3×4 Byte 数据域 + 2 Byte CRC16 = 21 Byte
- 发送总 bit 数：21 Byte × (1+8+1) = 210 bit
- 发送耗时为：210 / 115200 = 0.00182s = 1.82ms
可以看到，仅仅三个 float32 数据就需要消耗这么长的时间。如果使用阻塞式发送是相当抽象的；使用 DMA 等缓存模式发送，则需要考虑：怎么才能够让上次发送完以后，下次立刻把新的数据塞入缓存中，从而使得数据接口使用率最高。
阅读友校的开源代码，发现目前大多数发送的逻辑都是：使用不同的定时器调度错开多次发送。然而这种做法在通讯需求不断增加时，无疑是压榨嵌入式开发的头发；如果用 RTOS 进行调用发送函数，又可能会造成进程间打架的情况；使用互斥保护就又变成阻塞了。
二、提出
从前面的背景看到，造成问题的原因是：发送速率在短时间内跟不上突发性的发送需求。因而这里抛砖引玉，提出一种用 FIFO 作为发送缓冲区的方法，从而优雅实现高效通讯。
每次发送时，需要所有在同一通讯链路上的设备都是用一个同样的发送管理函数。这个发送函数需要做如下的事情：
- 判断通讯层是否空闲：空闲，就直接发送。不空闲进入下一步
- 通讯层不空闲：先存入缓存队列，等待重发调用
- 定时器或发送结束中断 不断唤起尝试发送：通过某种优先级或调度机制，判断定时器触发内是否能发送，不断尝试
这种发送函数的优点如下：
- 能够实现缓冲，应对大量数据的顺时浪潮
- 能够实现失败重发，重发次数管理等功能。
- 可以对没有仲裁功能的半双工的总线进行收发管理，防止总线同时被占用导致撞车（说的就是你，垃圾 485IC 配 低速 485 总线）
以下是我的一个伪代码（看个意思就好）APP_MODBUS_RETURN_T AppModbus_Transmit(AppModbus_modbus_t *modbus,
uint8_t address, uint8_t funCode, uint8_t *pData, uint16_t dataSize)
{
APP_MODBUS_RETURN_T ret = APP_MODBUS_OK;
// 开始尝试发送
if (modbus->status.stateCode == APP_MODBUS_STATE_READY) // 判断总线层状态
{
if (modbus->parameter.ctx.getComStatus(modbus->parameter.comHandle)
== 0) // 获取通讯层的状态，看看能不能直接发送
{
// 能直接发送，数据打包好直接发
uint16_t sizeEnpacked = 4 + dataSize;
ProtocalEnpack(modbus, address, funCode, pData, dataSize);
modbus->parameter.ctx.transmitFunction(modbus->parameter.comHandle,
modbus->parameter.pTxBuffer, sizeEnpacked);
}
else // 不能直接发，那就放到队列里慢慢发
{
if (!AppFifo_IsFull(modbus->parameter.txQueue)) // Q 没满，可以塞进去
{
// 数据打包，方便后面重发
AppModbus_modbusTransFrame_t transPack;
transPack.addressField = address;
transPack.functionCodeField = funCode;
// 找个地方存好源数据，防止在要发送前就被销毁了
transPack.pDataField = malloc(dataSize);
memcpy(transPack.pDataField, pData, dataSize);
transPack.dataFeildSize = dataSize;
// 发送包里面存放了重发相关的参数
transPack.TimeoutTick = modbus->parameter.rxMaximumTimeoutTick;
transPack.txRetry = modbus->parameter.txMaximumRetry;
// 执行发送
AppFifo_Add(modbus->parameter.txQueue, &transPack);
}
}
}
else
ret = (APP_MODBUS_RETURN_T) modbus->status.stateCode;
return ret;
}
在发送的调度器中，则需要自行处理好重发间隔、重发次数或消息发送优先级等额外功能。由于文章字节限制，这贴不下代码了，只好作罢。不过这部分和实际需求结合比较紧密，且没什么难度，实现起来还是很简单的。
三、可能遇到的问题
1. malloc 与 RTOS
上面的代码中，可以看到，数据被放入 FIFO 时，会另外开一片内存先把要发的数据存起来。以我个人为例，我的通讯是基于 38400 波特率的 485 总线进行的，发送一串数据要很久。而调用这段发送函数后，可能原来的数据是函数中间的临时变量，没发完就被销毁了。所以需要 malloc 存起来。并在发送完成时，要及时 free 掉这片空间。例如如下伪代码：
```
if (QueueTransmit(AppModbus_Lut) == APP_MODBUS_OK) // 成功发送
{
free(nextFrame.pDataField); // 发完了，释放内存
AppFifo_Discard(AppModbus_Lut->parameter.txQueue, 1, E_FIFO_FRONT); // 把这一帧丢掉
}
```
当 FreeRTOS 时，则需要尤其注意：如果你在**中断**中需要对内存管理，不要使用 **pvPortMalloc 和 vPortFree**！因为 FreeRTOS 规定只有带有 ISR 的函数才能在中断里被调用。
2. FIFO 的改进
我是用的 FIFO 开源（见文末）是只有入队、出队等基本操作的。但如果涉及到重发延时、重发次数管理等，可能需要把队首的元素拿出来看一眼（peek），或者修改其中的某些参数后立刻把它塞回原位（注意是原位而不是重新入队），才能方便的判断其是否到达某种临界条件。因而如果你使用的 FIFO 不包含相关操作，则需要自己编写函数。或者不使用定时器不断轮询，而是采用计时中断的方法。
我个人觉得定时器定时触发某个通讯调度器的方法比较优雅而且好移植，所以给原开源增加了两个函数：fifo_peek 和 fifo_paste，分别对应上面说到的两种需求。
四、参考资料
FIFO 队列开源连接：https://github.com/geekfactory/FIFO
Tips: 阅读前请自动 #include <stdint.h>
Tips2: 本文写自某老灯，希望能为 RM 的发展做出一点微薄的贡献，让大家不要再为一些底层的琐事困扰。
Tips3: 本人水平有限，能力不强，仅作为抛砖引玉，提出一种可能的方案，大佬轻喷。
一、背景
使用某种通讯接口进行发送是单片机开发最常见的场景之一。然而通讯速率不同会影响发送策略。
例如当通讯接口速率较快、且发送数据较少时（如USBFS），发送可以看作瞬间完成，反映到代码中：可以认为调用完这次发送代码后，便可以立刻再次调用下一次发送；CPU 下次访问发送寄存器/缓存之前，寄存器/缓存已经清空。也就是连着写几行发送也没有什么大问题。
但如果不幸的是我们在使用某种速率较慢的接口，且一次要发送大量数据时，发送就会遇到大问题。例如使用波特率 115200、8 数据位、1 停止位、无校验位的串口，发送 3 个 float32 类型的数据。帧格式仿照某比赛使用的“串口协议手册”中的格式，我们可以计算出：
- 发送总 Byte 数：5 Byte 帧头 + 2 Byte cmdId + 3×4 Byte 数据域 + 2 Byte CRC16 = 21 Byte
- 发送总 bit 数：21 Byte × (1+8+1) = 210 bit
- 发送耗时为：210 / 115200 = 0.00182s = 1.82ms
可以看到，仅仅三个 float32 数据就需要消耗这么长的时间。如果使用阻塞式发送是相当抽象的；使用 DMA 等缓存模式发送，则需要考虑：怎么才能够让上次发送完以后，下次立刻把新的数据塞入缓存中，从而使得数据接口使用率最高。
阅读友校的开源代码，发现目前大多数发送的逻辑都是：使用不同的定时器调度错开多次发送。然而这种做法在通讯需求不断增加时，无疑是压榨嵌入式开发的头发；如果用 RTOS 进行调用发送函数，又可能会造成进程间打架的情况；使用互斥保护就又变成阻塞了。
二、提出
从前面的背景看到，造成问题的原因是：发送速率在短时间内跟不上突发性的发送需求。因而这里抛砖引玉，提出一种用 FIFO 作为发送缓冲区的方法，从而优雅实现高效通讯。
每次发送时，需要所有在同一通讯链路上的设备都是用一个同样的发送管理函数。这个发送函数需要做如下的事情：
- 判断通讯层是否空闲：空闲，就直接发送。不空闲进入下一步
- 通讯层不空闲：先存入缓存队列，等待重发调用
- 定时器或发送结束中断 不断唤起尝试发送：通过某种优先级或调度机制，判断定时器触发内是否能发送，不断尝试
这种发送函数的优点如下：
- 能够实现缓冲，应对大量数据的顺时浪潮
- 能够实现失败重发，重发次数管理等功能。
- 可以对没有仲裁功能的半双工的总线进行收发管理，防止总线同时被占用导致撞车（说的就是你，垃圾 485IC 配 低速 485 总线）
以下是我的一个伪代码（看个意思就好）APP_MODBUS_RETURN_T AppModbus_Transmit(AppModbus_modbus_t *modbus,
uint8_t address, uint8_t funCode, uint8_t *pData, uint16_t dataSize)
{
APP_MODBUS_RETURN_T ret = APP_MODBUS_OK;
// 开始尝试发送
if (modbus->status.stateCode == APP_MODBUS_STATE_READY) // 判断总线层状态
{
if (modbus->parameter.ctx.getComStatus(modbus->parameter.comHandle)
== 0) // 获取通讯层的状态，看看能不能直接发送
{
// 能直接发送，数据打包好直接发
uint16_t sizeEnpacked = 4 + dataSize;
ProtocalEnpack(modbus, address, funCode, pData, dataSize);
modbus->parameter.ctx.transmitFunction(modbus->parameter.comHandle,
modbus->parameter.pTxBuffer, sizeEnpacked);
}
else // 不能直接发，那就放到队列里慢慢发
{
if (!AppFifo_IsFull(modbus->parameter.txQueue)) // Q 没满，可以塞进去
{
// 数据打包，方便后面重发
AppModbus_modbusTransFrame_t transPack;
transPack.addressField = address;
transPack.functionCodeField = funCode;
// 找个地方存好源数据，防止在要发送前就被销毁了
transPack.pDataField = malloc(dataSize);
memcpy(transPack.pDataField, pData, dataSize);
transPack.dataFeildSize = dataSize;
// 发送包里面存放了重发相关的参数
transPack.TimeoutTick = modbus->parameter.rxMaximumTimeoutTick;
transPack.txRetry = modbus->parameter.txMaximumRetry;
// 执行发送
AppFifo_Add(modbus->parameter.txQueue, &transPack);
}
}
}
else
ret = (APP_MODBUS_RETURN_T) modbus->status.stateCode;
return ret;
}
在发送的调度器中，则需要自行处理好重发间隔、重发次数或消息发送优先级等额外功能。由于文章字节限制，这贴不下代码了，只好作罢。不过这部分和实际需求结合比较紧密，且没什么难度，实现起来还是很简单的。
三、可能遇到的问题
1. malloc 与 RTOS
上面的代码中，可以看到，数据被放入 FIFO 时，会另外开一片内存先把要发的数据存起来。以我个人为例，我的通讯是基于 38400 波特率的 485 总线进行的，发送一串数据要很久。而调用这段发送函数后，可能原来的数据是函数中间的临时变量，没发完就被销毁了。所以需要 malloc 存起来。并在发送完成时，要及时 free 掉这片空间。例如如下伪代码：
```
if (QueueTransmit(AppModbus_Lut) == APP_MODBUS_OK) // 成功发送
{
free(nextFrame.pDataField); // 发完了，释放内存
AppFifo_Discard(AppModbus_Lut->parameter.txQueue, 1, E_FIFO_FRONT); // 把这一帧丢掉
}
```
当 FreeRTOS 时，则需要尤其注意：如果你在**中断**中需要对内存管理，不要使用 **pvPortMalloc 和 vPortFree**！因为 FreeRTOS 规定只有带有 ISR 的函数才能在中断里被调用。
2. FIFO 的改进
我是用的 FIFO 开源（见文末）是只有入队、出队等基本操作的。但如果涉及到重发延时、重发次数管理等，可能需要把队首的元素拿出来看一眼（peek），或者修改其中的某些参数后立刻把它塞回原位（注意是原位而不是重新入队），才能方便的判断其是否到达某种临界条件。因而如果你使用的 FIFO 不包含相关操作，则需要自己编写函数。或者不使用定时器不断轮询，而是采用计时中断的方法。
我个人觉得定时器定时触发某个通讯调度器的方法比较优雅而且好移植，所以给原开源增加了两个函数：fifo_peek 和 fifo_paste，分别对应上面说到的两种需求。
四、参考资料
FIFO 队列开源连接：https://github.com/geekfactory/FIFO
请问这篇文章对你有用吗？
【分享帖】使用 FIFO 优化发送程序
