WhyCan Forum(哇酷开发者社区)

简易的read方法

Linux遵循Unix“一切皆文件”的设计哲学，将串口抽象为一个设备文件。读取串口和读取文件的流程一样，先通过open系统调用打开串口，然后通过read系统调用读取：

#define FRAME_SIZE 8
unsigned char buf[FRAME_SIZE];
int sensorFd = open(sensorPath, O_RDWR | O_NOCTTY);
read(sensorFd, &buf, FRAME_SIZE);

open默认以阻塞方式打开文件，因此read会使线程挂起，直到串口接收到可用的数据。read可以通过第二个参数count设置读取到的字节数，我们把count设为一帧的长度，希望read会阻塞线程到串口接收完一帧为止。但read的行为果真如此吗？阅读read的manual，其对第二个参数count的描述是“最多读取的字节数”，也就是说，read不一定会读取到count个字节。实际上，对于串口的情形，默认只要有一个字节被接收到，read就会立即返回，其返回值是实际接收到的字节数。因此，通过设置count来保证read能读完一帧是不可行的。
于是我们在第二次尝试中让read循环读取一个字节，然后把这个字节写到一个buffer中，直到读取了一个完整的帧：[2]

ssize_t read2(int fd, void *buf, size_t count) {
    ssize_t read_length = 0;
    while (read_length < count) {
        ssize_t delta = read(fd, buf + read_length, count - read_length);
        if (delta == -1) return -1;
        read_length += delta;
    }
    return read_length;
}

这个方法是可行的，但并非完美的：read是一个系统调用，执行时会导致内核陷入，因此开销比一般的过程调用要更加高昂。当然，因为进行阻塞的read时，执行的频率等于串口的波特率，而串口的波特率相对于处理器主频而言一般是很低的，这里系统调用的overhead不一定会造成性能瓶颈。但我们还是要问，比起每次读取一字节，有没有更好的方式呢？

实时性再探

借助termios的力量，我们成功让读取一帧的时间开销从线性（对帧长）倍减到了常数，可喜可贺！于是我们可以收工睡觉去了吗？且慢，再设想一下我们的程序被系统调度起来的情形。虽然Linux内核是非实时的，但尤其是当帧率不高时，可以预见有不小的几率我们的程序能及时响应连续的两帧，这时前述方法的一个问题显现了出来：由于每次读取（大概率）没有和帧对齐，我们从接收数据尾部搜寻到一帧时，实际上就把之后的数据，也就是下一帧的前半截丢掉了！我们当然可以在每次读取时加一个判断，确保搜寻到的是完整的一帧，但当连续的两帧原本都能被及时处理时，后一帧却被浪费了，他的前半截被上一次处理丢弃，后半截则被这一次处理无视。这会对我们应用的实时性造成一定影响，尤其是当帧率不高而使每一帧都比较重要时。解决办法是显而易见的：加一个生命周期在每帧处理函数之外的缓存以保存后一帧的前部，然后在后一次处理时取出来。但不那么显而易见的是，这个缓存该如何实现。

最粗暴的办法是直接把后一帧的前部拷贝到那个外部的缓存里，下次处理时首先就把缓存里的内容拷贝回来。不过当帧较长时，拷贝的开销就比较可观了，这不是一个scalable的方法。一些人可能马上会想到适于这个工作的一种数据结构：环形缓存。每次处理直接把数据读取到一个大小为帧长的环形缓存里，下一次读取到的帧的尾部会自动和残留在缓存里的前部拼接起来。理想是美好的，然而read并没有提供针对环形缓存的接口，因此在一次读取多个字节时可能发生越界。于是我们又面临一个两难的抉择：要么舍弃来之不易的一次读取多个字节的能力，让read逐字节地写入到环形缓存上，要么再引入一个复杂的中间层以对接read和环形缓存。

我没有太多数据结构相关的知识，针对这个困难，我姑且提出一种折衷的办法，来平衡复杂度、实时性和开销。它不一定是最好的方案，仅供大家讨论：把之前4kB的buffer再扩大一点（好吧，我个人的选择是扩大到8k，似乎不能说是“一点”了），然后每次读取直接写入到buffer上次读取的数据的末尾。这样一直写下去当然是要越界的，于是在每次读取前加一个判断，如果缓存剩余的空间已不足4k（为什么是4k？因为为了清空串口缓冲区，每次可能读取的最多字节数就是4k），就把此次读取的位置指向buffer的头部。这时后一帧的前部还是被丢掉了，但可以想见，把buffer设定得越大，这种情况发生的频率就越低，实际上是一个空间开销对实时性的tradeoff：

unsigned char buffer[8192];
unsigned char *buftop = buffer + 8192;
unsigned char *bufesp = buffer;

const unsigned char* getFrame(){
    if(bufesp + 4096 > buftop){
        bufesp = buffer;
    }
    bufesp += read(_sensorFd, bufesp, 4096);
    return serachFrame(bufesp);
}

考虑到简洁以及不同协议帧格式的差异，这里搜寻帧的操作用一行伪码serachFrame()代替。我在自己的应用中对上述机制的完整实现发布在了GitHub上：

https://github.com/SdtElectronics/yarn-fw/blob/master/src/drivers/lStream.cpp

楼主不是计科专业出身，以上内容均为Web搜索、查询手册加上个人的一些胡思乱想得来的。如果有什么贻笑大方的地方，还请坛友不吝赐教。

参考资料：

[1]: https://stackoverflow.com/a/66740261/10627291

[2]: https://stackoverflow.com/a/32632249/10627291

[3]: https://www.cmrr.umn.edu/~strupp/serial.html#4_2

最近编辑记录 SdtElectronics (2021-03-25 21:54:02)

虽然方法很山寨，但也是动过脑筋的干货。