C - select() 似乎阻塞的时间超过了超时

Question

我正在编写一个数据采集程序，需要

• 使用 select() 等待串行

• 读取串行数据（RS232 115200 波特），

• 给它加时间戳（clock_gettime()），

• 在 SPI 上读取 ADC，

• 解释一下，

• 通过另一个 tty 设备发送新数据

• 循环和重复

ADC 暂时无关紧要。

在循环结束时，我再次使用 select() 以 0 超时轮询并查看数据是否已经可用，如果是，则表示我已经溢出，即我希望循环在更多数据之前结束，并且循环开始时的 select() 会阻塞并在它到达时立即获取它。

数据应该每 5 毫秒到达一次，我的第一个 select() 超时计算为（5.5 毫秒 - 循环时间） - 应该是大约 4 毫秒。

我没有超时，但有很多超限。

检查时间戳会发现 select() 阻塞的时间超过了超时时间（但仍然返回 >0）。 看起来 select() 在超时前获取数据后返回较晚。

这可能在 1000 次重复中发生 20 次。 可能是什么原因？我该如何解决？

编辑： 这是代码的精简版本（我做了比这更多的错误检查！）

#include <bcm2835.h> /* for bcm2835_init(), bcm2835_close() */

int main(int argc, char **argv){

    int err = 0;

    /* Set real time priority SCHED_FIFO */
    struct sched_param sp;
    sp.sched_priority = 30;
    if ( pthread_setschedparam(pthread_self(), SCHED_FIFO, &sp) ){
        perror("pthread_setschedparam():");
        err = 1;
    }

    /* 5ms between samples on /dev/ttyUSB0 */
    int interval = 5;

    /* Setup tty devices with termios, both totally uncooked, 8 bit, odd parity, 1 stop bit, 115200baud */
    int fd_wc=setup_serial("/dev/ttyAMA0");
    int fd_sc=setup_serial("/dev/ttyUSB0");

    /* Setup GPIO for SPI, SPI mode, clock is ~1MHz which equates to more than 50ksps */
    bcm2835_init();
    setup_mcp3201spi();

    int collecting = 1;

    struct timespec starttime;
    struct timespec time;
    struct timespec ftime;
    ftime.tv_nsec = 0;

    fd_set readfds;
    int countfd;
    struct timeval interval_timeout;
    struct timeval notime;

    uint16_t p1;
    float w1;

    uint8_t *datap = malloc(8);
    int data_size;
    char output[25];

    clock_gettime(CLOCK_MONOTONIC, &starttime);

    while ( !err && collecting ){   
        /* Set timeout to (5*1.2)ms - (looptime)ms, or 0 if looptime was longer than (5*1.2)ms */
        interval_timeout.tv_sec = 0;
        interval_timeout.tv_usec = interval * 1200 - ftime.tv_nsec / 1000;
        interval_timeout.tv_usec = (interval_timeout.tv_usec < 0)? 0 : interval_timeout.tv_usec;
        FD_ZERO(&readfds);
        FD_SET(fd_wc, &readfds);    
        FD_SET(0, &readfds); /* so that we can quit, code not included */   
        if ( (countfd=select(fd_wc+1, &readfds, NULL, NULL, &interval_timeout))<0 ){
            perror("select()");
            err = 1;
        } else if (countfd == 0){
            printf("Timeout on select()\n");
            fflush(stdout);
            err = 1;
        } else if (FD_ISSET(fd_wc, &readfds)){
            /* timestamp for when data is just available */
            clock_gettime(CLOCK_MONOTONIC, &time)
            if (starttime.tv_nsec > time.tv_nsec){
                time.tv_nsec = 1000000000 + time.tv_nsec - starttime.tv_nsec;
                time.tv_sec = time.tv_sec - starttime.tv_sec - 1;
            } else {
                time.tv_nsec = time.tv_nsec - starttime.tv_nsec;
                time.tv_sec = time.tv_sec - starttime.tv_sec;
            }

            /* get ADC value, which is sampled fast so corresponds to timestamp */
            p1 = getADCvalue();

            /* receive_frame, receiving is slower so do it after getting ADC value. It is timestamped anyway */
            /* This function consists of a loop that gets data from serial 1 byte at a time until a 'frame' is collected. */
            /* it uses select() with a very short timeout (enough for 1 byte at baudrate) just to check comms are still going */
            /* It never times out and behaves well */
            /* The interval_timeout is passed because it is used as a timeout for responding an ACK to the device */
            /* That select also never times out */
            ireceive_frame(&datap, fd_wc, &data_size, interval_timeout.tv_sec, interval_timeout.tv_usec);

            /* do stuff with it */
            /* This takes most of the time in the loop, about 1.3ms at 115200 baud */
            snprintf(output, 24, "%d.%04d,%d,%.2f\n", time.tv_sec, time.tv_nsec/100000, pressure, w1);
            write(fd_sc, output, strnlen(output, 23)); 

            /* Check how long the loop took (minus the polling select() that follows */ 
            clock_gettime(CLOCK_MONOTONIC, &ftime);
            if ((time.tv_nsec+starttime.tv_nsec) > ftime.tv_nsec){
                ftime.tv_nsec = 1000000000 + ftime.tv_nsec - time.tv_nsec - starttime.tv_nsec;
                ftime.tv_sec = ftime.tv_sec - time.tv_sec - starttime.tv_sec - 1;
            } else {
                ftime.tv_nsec = ftime.tv_nsec - time.tv_nsec - starttime.tv_nsec;
                ftime.tv_sec = ftime.tv_sec - time.tv_sec - starttime.tv_sec; 
            }

            /* Poll with 0 timeout to check that data hasn't arrived before we're ready yet */
            FD_ZERO(&readfds);
            FD_SET(fd_wc, &readfds);
            notime.tv_sec = 0;  
            notime.tv_usec = 0; 
            if ( !err && ( (countfd=select(fd_wc+1, &readfds, NULL, NULL, &notime)) < 0 )){
                perror("select()");
                err = 1;
            } else if (countfd > 0){
                printf("OVERRUN!\n");
                snprintf(output, 25, ",,,%d.%04d\n\n", ftime.tv_sec, ftime.tv_nsec/100000);
                write(fd_sc, output, strnlen(output, 24)); 
            }

        }

    }


    return 0;

}

我在输出的串行流上看到的时间戳是相当规则的（通常下一个循环会赶上偏差）。输出的 sn-p：

6.1810,0,225.25
6.1867,0,225.25
6.1922,0,225.25
6,2063,0,225.25
,,,0.0010

在这里，最多 6.1922 秒一切正常。下一个样本是 6.2063 - 上一个样本之后的 14.1 毫秒，但它没有超时，6.1922-6.2063 的前一个循环也没有通过轮询 select() 捕获溢出。我的结论是最后一个循环是采样时间，并且选择在没有超时的情况下花费了 -10ms 太长的返回时间。

,,,0.0010 表示循环之后的循环时间 (ftime) - 我真的应该检查出错时的循环时间。我明天试试。

Answer 1

传递给select 的超时时间是一个粗略的下限——select 允许延迟你的进程稍微多一点。特别是，如果您的进程被不同的进程（上下文切换）或内核中的中断处理抢占，您的进程将被延迟。

以下是 Linux 手册页关于该主题的内容：

注意超时时间会向上取整到系统时钟 粒度和内核调度延迟意味着阻塞 间隔可能会超出少量。

这是 POSIX 标准：

实现可能 还对超时间隔的粒度进行了限制。如果 请求的超时间隔需要比 实现支持，实际超时间隔为 向上舍入到下一个支持的值。

在通用系统上很难避免这种情况。通过将进程锁定在内存中（mlockall）并将进程设置为实时优先级（使用sched_setscheduler 和SCHED_FIFO，并记住经常睡觉，您将获得合理的结果，尤其是在多核系统上足以让其他进程有机会运行）。

一种更困难的方法是使用专用于运行实时代码的实时微控制器。有些人声称reliably sample at 20MHz on fairly cheap hardware 使用了这种技术。

Answer 2

如果struct timeval 的值设置为零，则select 不会阻塞，但如果超时参数是空指针，它会...

如果 timeout 参数不是一个 NULL 指针，它指向一个 struct timeval 类型的对象，它指定了一个最大间隔 等待选择完成。如果超时参数指向 一个成员为 0 的 struct timeval 类型的对象，select() 确实 不阻塞。如果 timeout 参数是一个 NULL 指针，select() 阻塞，直到事件导致其中一个掩码返回 一个有效的（非零）值或直到出现需要的信号 发表。如果时间限制在任何事件发生之前到期 将导致掩码之一设置为非零值，select() 成功完成并返回 0。

阅读更多here

EDIT 解决 cmets，并添加新信息：

有几点值得注意。

首先 - 在 cmets 中，建议将 sleep() 添加到您的工作循环中。这是一个很好的建议。 stated here 的原因虽然处理了线程入口点，但仍然适用，因为您正在实例化一个连续循环。

第二 - Linux select() 是一个具有有趣实现历史的系统调用，因此在实现之间具有一系列不同的行为，其中一些可能助长您所看到的意外行为。我不确定 Linux Arch Linux 的主要血统来自哪个，但 man7.org page for select() 包括以下两个部分，根据您的描述似乎描述了可能导致您遇到的延迟的情况。

校验和错误：

Under Linux, select() may report a socket file descriptor as "ready   
for reading", while nevertheless a subsequent read blocks.  This could  
for example happen when data has arrived but upon examination has wrong  
checksum and is discarded.  

竞态条件：（介绍和讨论pselect()）

...Suppose the signal handler sets a global flag and returns.  Then a test  
of this global flag followed by a call of select() could hang indefinitely  
if the signal arrived just after the test but just before the call...   

根据您对观察结果的描述，并根据您的 Linux 版本的实现方式，这些实现中的任何一个功能都可能是贡献者。