楼主最近的一个项目需要在Linux下用串口接收来自一个传感器的数据流，本以为很简单的任务在探究一番之后却发现有不少名堂，特把自己的一些心得记录下来和大家分享。
在广泛被采用的传感器与上位机的接口中，串口是因缺少I2C、SPI等协议所有的硬帧同步机制，而显得较为特殊的一种。这类传感器（或其他传输数据有Burst特性的设备）发送的数据流一般有如下特点：
传输的突发性
传感器不连续按波特率发送数据，而是按某一周期重复发送，或在某事件触发时发送数据
数据的时效性
应用只关心传感器最新发送的一个或数个数据，此前的数据因已过时而不再重要了
数据帧格式固定
传感器产生的数据帧有固定的格式，通常可能包括起始字节，中止字节，校验和，帧的长度也一般是固定的；但是数据载荷可能是二进制格式，从而可能正好为起始字节或终止字节的值，由此造成帧对齐的误判
这些特性让数据帧的解析变得复杂。下面从比较简单的接收方法开始分析，看看会遇到什么问题。
离线
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不一定会造成性能瓶颈。但我们还是要问，比起每次读取一字节，有没有更好的方式呢？
离线
上述的性能问题的实质是，每次read读取的数据量太小，使得读完一帧的开销变大。我们希望有控制使read阻塞的字节数的方法，而之前提到的count参数并不能做到这点。操作系统的设计者当然对此有所考虑，在Linux部分实现的POSIX规范中就通过termios提供了对于串口的更为细化的控制。
termios将对串口的读写分为两类：Canonical Mode和Raw Mode。历史上，串口被广泛用于和终端交互，Canonical Mode就是提供终端的命令发送和回显功能的模式。这种模式下，termios通过预先规定的分隔符将输入分隔为行，这样可以使read阻塞到收完一行为止。我们发现似乎可以利用这个模式来满足我们的需求：将标志包尾的字节设为分隔符，内核就会让read阻塞到读完一帧为止。问题解决了吗？再仔细审视一下这个方法，我们会发现仍然有许多问题：
分隔符的出现不一定标志着读完了一帧。因为没有控制read读取的最小字节数，读取到的可能是某一帧的尾部，或是数据载荷中的某一字节恰好取到了分隔符的值，导致对帧结尾的误判
这种机制并不能保证我们读到的是最新的数据，可能串口的缓冲区里已经积压了太多的过时的帧，而read总是从最早接收的字节开始读取的
阅读termios的文档可知，Canonical Mode还规定了一些控制字符，这意味着数据载荷取到ASCII可打印字符以外的值时，可能会产生意料外的结果
不管怎样，Canonical Mode至少解决了频繁的系统调用带来的性能问题，却同时带来了更大的麻烦。你大可以通过调整termios的各种设置以及加入更多的条件判断来解决这些问题，但不应忘了简洁是程序设计的最高原则之一。既然利用Canonical Mode有如此诸多麻烦，不妨转而了解一下termios提供的另一个模式，Raw Mode。
Raw Mode是termios专为二进制数据流制定的模式，并且通过一个宏VMIN实现了我们之前苦苦追寻的对read读取最小字节数的控制。现在可以通过在程序中设置termios结构体来确保能一次读完一帧了：[3]
struct termios sssConf;
//get the current options
tcgetattr(_sensorFd, &sssConf);
//set raw input, 1 second timeout
cfmakeraw(&sssConf);
sssConf.c_cc[VMIN] = FRAME_SIZE;
sssConf.c_cc[VTIME] = 10;
cfsetispeed(&sssConf, B19200);
//set the options
tcsetattr(_sensorFd, TCSANOW, &sssConf);但这才是解决问题的第一步，上面的代码既不能保证帧头和buffer的首字节是对齐的，也不能保证读取到的帧是最新的。对于前一个问题，我们可以通过搜寻帧的起始或终止字节来对齐一帧，并通过包的校验和来验证帧的有效性。对于后一个问题，我们可以把read的count设为串口缓冲区的大小，从而获取到最新的数据帧，并从read到的数据末尾开始搜寻一帧。这个方法看起来已经足够好了，除了buffer似乎有点大——4kB，也即内核默认的串口缓冲区大小。实际上内核实现的串口驱动提供了一个ioctl的request来控制串口缓冲区的大小，但不同版本的内核上ioctl的语义无法保证一致；另一方面，4kB的buffer和内核给一个线程设定的默认栈空间上限——8M相比，似乎又显得微不足道了，因此在内存资源并不紧张时，优化buffer大小显得并无必要。
离线
借助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)
离线