Perl：读取fifos非阻塞

我现在已经制作了第二个版本来缓冲内存中的一行.

它工作但它需要所有fifos在它开始之前连接.这有效：

window1$mkfifo {1..100}
window1$parcat {1..100} | pv >/dev/null

window2$parallel -j0 'cat bigfile > ' ::: *

这不会给出任何输出(因为100没有连接)：

window1$mkfifo {1..100}
window1$parcat {1..100} | pv >/dev/null

window2$parallel -j0 'cat bigfile > ' ::: {1..99}

我尝试使用open’<'.这解决了上述问题,但现在它并没有停止在EOF. 我怎么做？ 最小版本(不支持大行而不退避等待)：

#!/usr/bin/perl

use Symbol qw(gensym);
use IPC::Open3;
use POSIX qw(:errno_h);
use Fcntl qw(:DEFAULT :flock);

for (@ARGV) {
    open($fh{$_},"<",$_) || die;
    # set fh non blocking($fh{$_});
    my $flags;
    fcntl($fh{$_},&F_GETFL,$flags) || die $!; # Get the current flags on the filehandle
    $flags |= &O_NONBLOCK; # Add non-blocking to the flags
    fcntl($fh{$_},&F_SETFL,$flags) || die $!; # Set the flags on the filehandle
}

while(keys %fh) {
    for(keys %fh) {
        my($string,$something_read) = non_blocking_read($_);
    print $string;
    }
    # Sleep 1 ms
    select(undef,undef,1/1000);
}

{
    my %buffer;

    sub non_blocking_read {

        my $file = shift;
        my $in = $fh{$file};
        my $rv = sysread($in,substr($buffer{$file},length $buffer{$file}),327680);
        if (!$rv) {
            if($! == EAGAIN) {
            # Would block: Nothing read
            return(undef,undef);
            } else {
            # This file is done
            close $in;
            delete $fh{$file};
            my $buf = $buffer{$file};
            delete $buffer{$file};
            return ($buf,1);
            }
        }

        # Find n for full line
        my $i = (rindex($buffer{$file},"n")+1);
        if($i) {
            # Return full line
            # Remove full line from $buffer
            return(substr($buffer{$file},$i),1,$i) = "");
        } else {
            # Something read,but not a full line
            return("",1);
        }
    }
}

完整版：重要的代码在前40行：其余的是经过良好测试的代码.

#!/usr/bin/perl

use Symbol qw(gensym);
use IPC::Open3;

for (@ARGV) {
    open($fh{$_},$_) || die;
    set_fh_non_blocking($fh{$_});
}

$ms = 1;
while(keys %fh) {
    for(keys %fh) {
    my($string,$something_read) = non_blocking_read($_);
    if($something_read) {
        $ms = 0.1;
        print $string;
    }
    }
    $ms = exp_usleep($ms);
}

{
    my %buffer;
    my $ms;

    sub non_blocking_read {
    use POSIX qw(:errno_h);

    my $file = shift;
    my $in = $fh{$file};
    my $rv = read($in,327680);
    if (!$rv) {
        if($! == EAGAIN) {
        # Would block: Nothing read
        return(undef,undef);
        } else {
        # This file is done
        close $in;
        delete $fh{$file};
        my $buf = $buffer{$file};
        delete $buffer{$file};
        return ($buf,1);
        }
    }

    #### Well-tested code below                                                           

    # Find n or r for full line
    my $i = (::rindex64($buffer{$file},"n")+1) ||
        (::rindex64($buffer{$file},"r")+1);
    if($i) {
        # Return full line
        # Remove full line from $buffer
        return(substr($buffer{$file},$i) = "");
    } else {
        # Something read,but not a full line
        return("",1);
    }
    }
}

sub rindex64 {
    # Do rindex on strings > 2GB.
    # rindex in Perl < v5.22 does not work for > 2GB
    # Input:
    #   as rindex except STR which must be passed as a reference
    # Output:
    #   as rindex
    my $ref = shift;
    my $match = shift;
    my $pos = shift;
    my $block_size = 2**31-1;
    my $strlen = length($$ref);
    # Default: search from end
    $pos = defined $pos ? $pos : $strlen;
    # No point in doing extra work if we don't need to.
    if($strlen < $block_size) {
    return rindex($$ref,$match,$pos);
    }

    my $matchlen = length($match);
    my $ret;
    my $offset = $pos - $block_size + $matchlen;
    if($offset < 0) {
    # The offset is less than a $block_size
    # Set the $offset to 0 and
    # Adjust block_size accordingly
    $block_size = $block_size + $offset;
    $offset = 0;
    }
    while($offset >= 0) {
    $ret = rindex(
        substr($$ref,$offset,$block_size),$match);
    if($ret != -1) {
        return $ret + $offset;
    }
    $offset -= ($block_size - $matchlen - 1);
    }
    return -1;
}

sub exp_usleep {
    # Sleep this many milliseconds.
    # Input:
    #   $ms = milliseconds to sleep
    # Returns:
    #   $ms + 10%
    my $ms = shift;
    select(undef,$ms/1000);
    return (($ms < 1000) ? ($ms * 1.1) : ($ms));
}

sub set_fh_non_blocking {
    # Set filehandle as non-blocking
    # Inputs:
    #   $fh = filehandle to be blocking
    # Returns:
    #   N/A
    my $fh = shift;
    $Global::use{"Fcntl"} ||= eval "use Fcntl qw(:DEFAULT :flock); 1;";
    my $flags;
    fcntl($fh,$flags) || die $!; # Get the current flags on the filehandle
    $flags |= &O_NONBLOCK; # Add non-blocking to the flags
    fcntl($fh,$flags) || die $!; # Set the flags on the filehandle
}