¿Por qué parece que pierdo datos con esta construcción de tubería bash?

Estoy tratando de combinar algunos programas como este (ignore cualquier inclusión adicional, este es un trabajo pesado en progreso):

pv -q -l -L 1  < input.csv | ./repeat <(nc "host" 1234)

Donde la fuente del programa de repetición se ve de la siguiente manera:

#include <fcntl.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

#include <iostream>
#include <string>

inline std::string readline(int fd, const size_t len, const char delim = '\n')
{
    std::string result;
    char c = 0;
    for(size_t i=0; i < len; i++)
    {
        const int read_result = read(fd, &c, sizeof(c));
        if(read_result != sizeof(c))
            break;
        else
        {
            result += c;
            if(c == delim)
                break;
        }
    }
    return result;
}

int main(int argc, char ** argv)
{
    constexpr int max_events = 10;

    const int fd_stdin = fileno(stdin);
    if (fd_stdin < 0)
    {
        std::cerr << "#Failed to setup standard input" << std::endl;
        return -1;
    }


    /* General poll setup */
    int epoll_fd = epoll_create1(0);
    if(epoll_fd == -1) perror("epoll_create1: ");
    {
        struct epoll_event event;
        event.events = EPOLLIN;
        event.data.fd = fd_stdin;
        const int result = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd_stdin, &event);
        if(result == -1) std::cerr << "epoll_ctl add for fd " << fd_stdin << " failed: " << strerror(errno) << std::endl;
    }

    if (argc > 1)
    {
        for (int i = 1; i < argc; i++)
        {
            const char * filename = argv[i];
            const int fd = open(filename, O_RDONLY);
            if (fd < 0)
                std::cerr << "#Error opening file " << filename << ": error #" << errno << ": " << strerror(errno) << std::endl;
            else
            {
                struct epoll_event event;
                event.events = EPOLLIN;
                event.data.fd = fd;
                const int result = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event);
                if(result == -1) std::cerr << "epoll_ctl add for fd " << fd << "(" << filename << ") failed: " << strerror(errno) << std::endl;
                else std::cerr << "Added fd " << fd << " (" << filename << ") to epoll!" << std::endl;
            }
        }
    }

    struct epoll_event events[max_events];
    while(int event_count = epoll_wait(epoll_fd, events, max_events, -1))
    {
        for (int i = 0; i < event_count; i++)
        {
            const std::string line = readline(events[i].data.fd, 512);                      
            if(line.length() > 0)
                std::cout << line << std::endl;
        }
    }
    return 0;
}

Me di cuenta de esto:

• Cuando solo uso la tubería ./repeat, todo funciona según lo previsto.
• Cuando solo uso la sustitución del proceso, todo funciona según lo previsto.
• Cuando encapsulo pv usando la sustitución de procesos, todo funciona según lo previsto.
• Sin embargo, cuando uso la construcción específica, ¡parece que pierdo datos (caracteres individuales) de stdin!

He probado lo siguiente:

• He tratado de deshabilitar el almacenamiento en búfer en la tubería entre pvy el ./repeatuso stdbuf -i0 -o0 -e0en todos los procesos, pero eso no parece funcionar.
• He cambiado epoll por encuesta, no funciona.
• Cuando miro el flujo entre pvy ./repeatcon tee stream.csv, esto parece correcto.
• Solía stracever lo que estaba sucediendo, y veo muchas lecturas de un solo byte (como se esperaba) y también muestran que faltan datos.

¿Me pregunto lo que está pasando? ¿O qué puedo hacer para investigar más?

Porque el nccomando dentro <(...)también leerá desde stdin.

Ejemplo más simple:

$ nc -l 9999 >/tmp/foo &
[1] 5659

$ echo text | cat <(nc -N localhost 9999) -
[1]+  Done                    nc -l 9999 > /tmp/foo

¿A dónde fue text? A través del netcat.

$ cat /tmp/foo
text

Su programa y nccompetir por el mismo stdin, y ncobtiene algo de él.

epoll () o poll () que regresan con E / POLLIN solo le indicarán que una sola lectura () puede no bloquearse.

No es que pueda hacer una lectura de un byte () hasta una nueva línea, como lo hace.

Digo mayo porque un read () después de epoll () devuelto con E / POLLIN aún puede bloquearse.

Su código también intentará leer EOF pasado e ignorará completamente cualquier error de lectura ().