c++ - 从 Linux 中的多个非阻塞命名管道读取

Question

基于位于 here 的类似示例在计算器中，我有三个命名管道，pipe_a、pipe_b 和 pipe_c，它们由外部进程提供。我想要一个读取器进程，无论写入这些管道中的任何内容，都可以输出到控制台。

下面的程序是一个一体化的 c 程序，它应该以非阻塞的方式读取三个管道，并在任何一个管道获取新数据时显示输出。

但是，它不起作用 - 它正在阻塞!如果pipe_a获取到数据，它会显示它，然后等待新数据到达pipe_b，等等...

select() 应该允许监视多个文件描述符，直到一个准备好，此时我们应该进入管道的读取函数并获取数据。

任何人都可以帮助确定为什么管道的行为就像它们处于阻塞模式一样吗？

/*
 * FIFO example using select.
 *
 * $ mkfifo /tmp/fifo
 * $ clang -Wall -o test ./test.c
 * $ ./test &
 * $ echo 'hello' > /tmp/fifo
 * $ echo 'hello world' > /tmp/fifo
 * $ killall test
 */

#include <sys/types.h>
#include <sys/select.h>
#include <errno.h>
#include <stdlib.h>
#include <stdio.h>
#include <fcntl.h>
#include <unistd.h>


// globals
int fd_a, fd_b, fd_c;
int nfd_a, nfd_b, nfd_c;
fd_set set_a, set_b, set_c;
char buffer_a[100*1024];
char buffer_b[100*1024];
char buffer_c[100*1024];


int readPipeA()
{
ssize_t bytes;
size_t total_bytes;

if (FD_ISSET(fd_a, &set_a)) {
    printf("\nDescriptor %d has new data to read.\n", fd_a);
    total_bytes = 0;
    for (;;) {
        printf("\nDropped into read loop\n");
        bytes = read(fd_a, buffer_a, sizeof(buffer_a));
        if (bytes > 0) {
            total_bytes += (size_t)bytes;
            printf("%s", buffer_a);
        } else {
            if (errno == EWOULDBLOCK) {
                printf("\ndone reading (%ul bytes)\n", total_bytes);
                break;
            } else {
                perror("read");
                return EXIT_FAILURE;
            }
        }
    }
}
}

int readPipeB()
{
ssize_t bytes;
size_t total_bytes;

if (FD_ISSET(fd_b, &set_b)) {
    printf("\nDescriptor %d has new data to read.\n", fd_b);
    total_bytes = 0;
    for (;;) {
        printf("\nDropped into read loop\n");
        bytes = read(fd_b, buffer_b, sizeof(buffer_b));
        if (bytes > 0) {
            total_bytes += (size_t)bytes;
            printf("%s", buffer_b);
        } else {
            if (errno == EWOULDBLOCK) {
                printf("\ndone reading (%ul bytes)\n", total_bytes);
                break;
            } else {
                perror("read");
                return EXIT_FAILURE;
            }
        }
    }
}
}

int readPipeC()
{
ssize_t bytes;
size_t total_bytes;

if (FD_ISSET(fd_c, &set_c)) {
    printf("\nDescriptor %d has new data to read.\n", fd_c);
    total_bytes = 0;
    for (;;) {
        printf("\nDropped into read loop\n");
        bytes = read(fd_c, buffer_c, sizeof(buffer_c));
        if (bytes > 0) {
            total_bytes += (size_t)bytes;
            printf("%s", buffer_c);
        } else {
            if (errno == EWOULDBLOCK) {
                printf("\ndone reading (%ul bytes)\n", total_bytes);
                break;
            } else {
                perror("read");
                return EXIT_FAILURE;
            }
        }
    }
}
}


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


    // create pipes to monitor (if they don't already exist)
    system("mkfifo /tmp/PIPE_A");
    system("mkfifo /tmp/PIPE_B");
    system("mkfifo /tmp/PIPE_C");


    // open file descriptors of named pipes to watch
    fd_a = open("/tmp/PIPE_A", O_RDWR | O_NONBLOCK);
    if (fd_a == -1) {
    perror("open");
    return EXIT_FAILURE;
    }
    FD_ZERO(&set_a);
    FD_SET(fd_a, &set_a);


    fd_b = open("/tmp/PIPE_B", O_RDWR | O_NONBLOCK);
    if (fd_b == -1) {
    perror("open");
    return EXIT_FAILURE;
    }
    FD_ZERO(&set_b);
    FD_SET(fd_b, &set_b);


    fd_c = open("/tmp/PIPE_C", O_RDWR | O_NONBLOCK);
    if (fd_c == -1) {
    perror("open");
    return EXIT_FAILURE;
    }
    FD_ZERO(&set_c);
    FD_SET(fd_c, &set_c);



    for(;;)
    {
        // check pipe A
        nfd_a= select(fd_a+1, &set_a, NULL, NULL, NULL);
        if (nfd_a) {
            if (nfd_a == -1) {
                perror("select");
                return EXIT_FAILURE;
            }
            readPipeA();
        }

        // check pipe B
        nfd_b= select(fd_b+1, &set_b, NULL, NULL, NULL);
        if (nfd_b) {
            if (nfd_b == -1) {
                perror("select");
                return EXIT_FAILURE;
            }
            readPipeB();
        }

        // check pipe C
        nfd_c= select(fd_c+1, &set_c, NULL, NULL, NULL);
        if (nfd_c) {
            if (nfd_c == -1) {
                perror("select");
                return EXIT_FAILURE;
            }
            readPipeC();
        }
    }

    return EXIT_SUCCESS;
}

---更新代码---

根据此处的反馈修改了应用程序，并阅读了更多内容:

    /*
     * FIFO example using select.
     *
     * $ mkfifo /tmp/fifo
     * $ clang -Wall -o test ./test.c
     * $ ./test &
     * $ echo 'hello' > /tmp/fifo
     * $ echo 'hello world' > /tmp/fifo
     * $ killall test
     */
    
    #include <sys/types.h>
    #include <sys/select.h>
    #include <sys/time.h>
    #include <sys/types.h>
    #include <errno.h>
    #include <stdlib.h>
    #include <stdio.h>
    #include <fcntl.h>
    #include <unistd.h>
    
    
    int readPipe(int fd)
    {
        ssize_t bytes;
        size_t total_bytes = 0;
        char buffer[100*1024];
    
        printf("\nDropped into read pipe\n");
        for(;;) {
            bytes = read(fd, buffer, sizeof(buffer));
            if (bytes > 0) {
                total_bytes += (size_t)bytes;
                printf("%s", buffer);
            } else {
                if (errno == EWOULDBLOCK) {
                    printf("\ndone reading (%d bytes)\n", (int)total_bytes);
                    break;
                } else {
                    perror("read");
                    return EXIT_FAILURE;
                }
            }
        }
        return EXIT_SUCCESS;
    }
    
    
    int main(int argc, char* argv[])
    {
        int fd_a, fd_b, fd_c;   // file descriptors for each pipe
        int nfd;                // select() return value
        fd_set read_fds;        // file descriptor read flags
        struct timeval tv;
        tv.tv_sec = 0;
        tv.tv_usec = 0;
    
        // create pipes to monitor (if they don't already exist)
        system("mkfifo /tmp/PIPE_A");
        system("mkfifo /tmp/PIPE_B");
        system("mkfifo /tmp/PIPE_C");
    
        // open file descriptors of named pipes to watch
        fd_a = open("/tmp/PIPE_A", O_RDWR | O_NONBLOCK);
        if (fd_a == -1) {
            perror("open");
            return EXIT_FAILURE;
        }
    
        fd_b = open("/tmp/PIPE_B", O_RDWR | O_NONBLOCK);
        if (fd_b == -1) {
            perror("open");
            return EXIT_FAILURE;
        }
    
        fd_c = open("/tmp/PIPE_C", O_RDWR | O_NONBLOCK);
        if (fd_c == -1) {
            perror("open");
            return EXIT_FAILURE;
        }
    
        FD_ZERO(&read_fds);
        FD_SET(fd_a, &read_fds);  // add pipe to the read descriptor watch list
        FD_SET(fd_b, &read_fds);
        FD_SET(fd_c, &read_fds);
    
        for(;;)
        {
            // check if there is new data in any of the pipes
            nfd = select(fd_a+1, &read_fds, NULL, NULL, &tv);
            if (nfd != 0) {
                if (nfd == -1) {
                    perror("select");
                    return EXIT_FAILURE;
                }
    
                if (FD_ISSET(fd_a, &read_fds)) {
                    readPipe(fd_a);
                }
            }
    
            nfd = select(fd_b+1, &read_fds, NULL, NULL, &tv);
            if (nfd != 0) {
                if (nfd == -1) {
                    perror("select");
                    return EXIT_FAILURE;
                }
    
                if (FD_ISSET(fd_b, &read_fds)){
                    readPipe(fd_b);
                }
            }
            nfd = select(fd_c+1, &read_fds, NULL, NULL, &tv);
            if (nfd != 0) {
                if (nfd == -1) {
                    perror("select");
                    return EXIT_FAILURE;
                }
                if (FD_ISSET(fd_c, &read_fds)){
                    readPipe(fd_c);
                }
            }
    
            usleep(10);
        }
        return EXIT_SUCCESS;
    }

当任何一个被监视的管道中有数据等待时，选择返回零 (0) 是否仍然存在问题？我一定没有正确使用 select() 和 fd_isset()。你能看出我做错了什么吗？谢谢。

Answer 1

The issue is that the select function is blocking. I understood select() to check flags to see if the read "would" block if it was performed, so that one can decide to perform the read or not. The pipe is being opened in RDWR and NONBLOCK mode.

你说问题是 select 函数被阻塞了，但是继续承认 NONBLOCK 标志只会让它成为 read 会阻塞。选择和阅读是两个不同的东西。

O_NONBLOCK 标志影响socket(因此，您的read 调用)；它不会改变 select 的行为，它有自己的超时/阻塞语义。

man select声明两个数字成员都设置为零的 timeout 参数会产生非阻塞轮询，而 NULL 的超时参数可能会导致无限期阻塞:

If the timeout parameter is a null pointer, then the call to pselect() or select() shall block indefinitely until at least one descriptor meets the specified criteria. To effect a poll, the timeout parameter should not be a null pointer, and should point to a zero-valued timespec timeval structure.

^{(注意。页面上方的文字表明，虽然 pselect() 采用 timespec 结构，但 select()采用 timeval 结构；我冒昧地将此逻辑应用于上述引用。)}

因此，在每个 select 调用之前构造一个 timeval，将其成员设置为零，并将其传递给 select。

一些注意事项，我们在这里:

1. 理想情况下，您只有 一个 select 调用，同时检查所有三个文件描述符，然后决定要读取 通过使用 fd_isset 检查您的 FD 集；

2. 我还建议在循环体的末尾放置一点 usleep，否则当数据不足时，您的程序会非常非常快地旋转。

   <