Linux命名管道上的read()超时

Question

假设我在Linux系统上创建了一个命名管道：

$ mkfifo my_pipe

接下来我想编写一个小型监控程序，该程序尝试从my_pipe中进行read()操作，但在一段时间后会超时。在以下伪代码中，我使用了一个虚构的函数wait_for_avail(fd, timeout_ms)：

int fd = open("my_pipe", O_RDONLY);
while (1) {
    //Fictional wait_for_avail(fd, timeout_ms). Is there a real function
    //that has this behaviour?
    int rc = wait_for_avail(fd, 500);
    if (rc == 1) {
        char buf[64];
        read(fd, buf, 64);
        //do something with buf
    } else {
        fprintf(stderr, "Timed out while reading from my_pipe\n");
        //do something else in the program
    }
}

我曾希望使用 POLLIN 标志的 poll 函数能够解决问题，但事实并非如此。通过我的简单试验，我发现它只是等待另一个进程以写入方式打开命名管道（而不是等到有数据可用，例如 read() 不会阻塞）。顺便提一下，由于某种原因，poll 忽略了超时设置，似乎会无限期地阻塞，直到另一个进程打开该管道为止。

我能想到的唯一其他解决方案是使用 O_NONBLOCK 选项来 open() 文件，并手动监控时间流逝，不断尝试读取 0 字节。

是否有更好的解决方案？

编辑：我这里的进程在打开命名管道时会阻塞。然而，如果您使用 O_NONBLOCK 选项，则文件会立即打开。此时，可以使用 poll() 函数等待（可选超时）管道的另一端被打开以进行写入。

然而，这仍然具有实现read()函数超时的行为。即使使用O_NONBLOCK打开管道，它仍然似乎在调用read()时被阻塞。

- Marco Merlini

---

不是read阻塞，而是open。 - undefined

以非阻塞模式打开，然后轮询等待超时或其他事件来打开FIFO的另一端。 - undefined

似乎poll不会等待字节可用。当与管道一起使用时，带有POLLIN的poll会等待管道的“另一端”被打开。对于普通文件，POLLIN总是立即返回。 - undefined

@Mahkoe，你没有读我的第一条评论。你的进程甚至无法通过open这一步。只有当有人打开管道的另一端时，open才会返回，而在open成功之前调用read是没有意义的。 - undefined

常规文件永远不会阻塞；读取操作将立即成功或失败。因此，在查询时，它始终处于“POLLIN”状态。 - undefined

Answer 1

你对于以非阻塞模式打开fifo的想法是正确的。如果你这样做，可以使用poll()/select()/等待另一端被打开，或者先超时。

下面的示例程序只会在无限循环中等待其他程序向my_pipe写入数据并回显所写的文本，有时会在没有数据或写入者时更新状态。

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <poll.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>

int main(void) {
  while (1) {
    int fd = open("my_pipe", O_RDONLY | O_NONBLOCK);
    if (fd < 0) {
      perror("open");
      return EXIT_FAILURE;
    }

    struct pollfd waiter = {.fd = fd, .events = POLLIN};

    while (1) {
      // 10 second timeout
      switch (poll(&waiter, 1, 10 * 1000)) {
      case 0:
        puts("The fifo timed out.");
        break;
      case 1:
        if (waiter.revents & POLLIN) {
          char buffer[BUFSIZ];
          ssize_t len = read(fd, buffer, sizeof buffer - 1);
          if (len < 0) {
            perror("read");
            return EXIT_FAILURE;
          }
          buffer[len] = '\0';
          printf("Read: %s\n", buffer);
        } else if (waiter.revents & POLLERR) {
          puts("Got a POLLERR");
          return EXIT_FAILURE;
        } else if (waiter.revents & POLLHUP) {
          // Writer closed its end
          goto closed;
        }
        break;
      default:
        perror("poll");
        return EXIT_FAILURE;
      }
    }
  closed:
    if (close(fd) < 0) {
      perror("close");
      return EXIT_FAILURE;
    }
  }
}

- Shawn

---

顺便说一下，似乎O_NONBLOCK对于读取命名管道的read()函数没有任何作用。请参考https://linux.die.net/man/7/fifo。 - undefined

@Shawn：当你尝试向一个已关闭的管道进行写入操作时，会触发SIGPIPE信号。据我所知，它从未在读取操作中触发过。 - undefined

@Shawn 你怎么知道管道中是否有可用的数据？听起来正是我所需要的。 - undefined

@Mahkoe 这就是 poll() 和相关函数所做的事情。 - undefined

@Mahkoe 如果管道关闭了，poll() 将立即返回 revents 中的 POLLHUP。然后 read() 返回 0，循环重复... 在 strace 下运行程序以查看（会有大量输出）。 - undefined

Answer 2

在得到@Shawn的大量帮助和耐心后，我设法找到了一个令我满意的答案。以下是名为pipe_watcher.c的文件内容：

#include <stdio.h>  //printf etc.
#include <errno.h>  //errno
#include <string.h> //perror
#include <signal.h> //SIGALRM, sigaction, sigset
#include <time.h>   //timer_create, timer_settime
#include <fcntl.h>  //open, O_RDONLY
#include <unistd.h> //close

/* This code demonstrates how you can monitor a named pipe with timeouts on the
 * read() system call.
 * 
 * Compile with:
 * 
 *  gcc -o pipe_watcher pipe_watcher.c -lrt
 * 
 * And run with:
 * 
 *  ./pipe_watcher PIPE_FILENAME
*/

//Just needed a dummy handler
void sigalrm_handler(int s) {
    return;
}

int main(int argc, char **argv) {
    //Check input argument count
    if (argc != 2) {
        puts("Usage:\n");
        puts("\t./pipe_watcher PIPE_FILENAME");
        return -1;
    }

    //Create a timer object
    timer_t clk;
    int rc = timer_create(CLOCK_REALTIME, NULL, &clk);
    if (rc < 0) {
        perror("Could not create CLOCK_REALTIME timer");
        return -1;
    }

    //Create some time values for use with timer_settime
    struct itimerspec half_second = {
        .it_interval = {.tv_sec = 0, .tv_nsec = 0},
        .it_value = {.tv_sec = 0, .tv_nsec = 500000000}
    };

    struct itimerspec stop_timer = {
        .it_interval = {.tv_sec = 0, .tv_nsec = 0},
        .it_value = {.tv_sec = 0, .tv_nsec = 0}
    };

    //Set up SIGALRM handler
    struct sigaction sigalrm_act = {
        .sa_handler = sigalrm_handler,
        .sa_flags = 0
    };
    sigemptyset(&sigalrm_act.sa_mask);
    rc = sigaction(SIGALRM, &sigalrm_act, NULL);
    if (rc < 0) {
        perror("Could not register signal handler");
        timer_delete(clk);
        return -1;
    }

    //We deliberately omit O_NONBLOCK, since we want blocking behaviour on
    //read(), and we're willing to tolerate dealing with the blocking open()
    int fd = open(argv[1], O_RDONLY);
    if (fd < 0) {
        char msg[80];
        sprintf(msg, "Could not open [%s]", argv[1]);
        perror(msg);
        timer_delete(clk);
        return -1;
    }

    puts("File opened");

    while (1) {
        //Buffer to read() into
        char buf[80];
        int len;

        //Set up a timer to interrupt the read() call after 0.5 seconds
        timer_settime(clk, 0, &half_second, NULL);

        //Issue read() system call
        len = read(fd, buf, 80);

        //Check for errors. The else-if checks for EOF
        if (len < 0) {
            if (errno == EINTR) {
                //This means we got interrupted by the timer; we can keep going
                fprintf(stderr, "Timeout, trying again\n");
                continue;
            } else {     
                //Something really bad happened. Time to quit.       
                perror("read() failed");
                //No point waiting for the timer anymore
                timer_settime(clk, 0, &stop_timer, NULL);
                break;
            }
        } else if (len == 0) {
            puts("Reached end of file");
            break;
        }

        //No error or EOF; stop the timer and print the results
        timer_settime(clk, 0, &stop_timer, NULL);
        write(STDOUT_FILENO, buf, len);
    }

    //Cleanup after ourselves
    timer_delete(clk);
    close(fd);
    return 0;
}

技巧是在（阻塞的）read()调用之前设置一个定时器。然后，我们只需检查read()的返回值，以查看是否由于超时而中断，是否发生了一般错误，是否达到了EOF，或者是否成功读取了数据。

只有一个小问题：你不能以非阻塞模式打开文件；这会导致open()在另一个进程打开管道进行写入之前一直阻塞。然而，在我的应用程序中，这实际上是一个可取的特性。你还可以设置SIGALRM来强制对open()设置超时，或者在另一个线程中完成。

事实上，这个技巧对于任何其他系统调用都适用，所以我可能会编写一个小助手库，使这种模式更容易使用。

编辑

还有一件事：在注册信号处理程序时，不要使用SA_RESTART标志非常重要。否则，即使系统调用被信号中断，Linux也会在信号处理完毕后再次尝试执行它。

- Marco Merlini