Building on a similar example located here in stackoverflow, I have three named pipes, pipe_a, pipe_b, and pipe_c that are being fed from external processes. I'd like to have a reader process that outputs to the console, whatever is written to any of these pipes.
The program below is an all-in-one c program that should read the three pipes in a non-blocking manner, and display output when any one of the pipes gets new data.
However, it isn't working - it is blocking! If pipe_a gets data, it will display it and then wait for new data to arrive in pipe_b, etc...
select() should allow the monitoring of multiple file descriptors until one is ready, at which time we should drop into the pipe's read function and get the data.
Can anyone help identify why the pipes are behaving like they are in blocking mode?
/*
* 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;
}
--- Updated Code ---
Modified the application based on the feedback here, and some more reading:
/*
* 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;
}
Still having an issue with the select returning zero (0) when there is data waiting in any one of the watched pipes? I must not be using the select() and fd_isset() correctly. Can you see what I'm doing wrong? Thanks.
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.
You say the problem is that the select function is blocking, but go on to admit that the NONBLOCK flag only makes it so that the read would block. Select and read are two different things.
The O_NONBLOCK flag affects the socket (and, consequently, your read calls); it does not change the behaviour of select, which has its own timeout/blocking semantics.
man select states that a timeout argument with both numeric members set to zero produces a non-blocking poll, whereas a timeout argument of NULL may lead to an indefinite block:
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
timespectimeval structure.
(NB. text further up the page indicates that, though pselect() takes a timespec structure, select() takes a timeval structure; I've taken the liberty of applying this logic to the above quotation.)
So, before each select call construct a timeval, set its members to zero, and pass that to select.
A couple of notes, while we're here:
Ideally you'd only have one select call, checking all three file descriptors at once, then deciding which pipes to read from by checking your FD set with fd_isset;
I also suggest putting a little usleep at the end of your loop body, otherwise your program is going to spin really, really quickly when starved of data.