并行读取 STDIN 时 $_ 为空

Question

我有一个遗留项目，它从 STDIN 获取大量数据并在 perl 脚本中逐行处理。行顺序并不重要。 这需要很长时间，所以我想并行进行。

经过一番研究，我发现Parallel::Loops 看起来很合适，但我无法让它工作，因为$_ 是空的。我的代码是：

#Initialize all vars etc

$pl->while ( sub { <STDIN> }, sub {
    print $_       # but $_ is empty
}

其他从 STDIN ir 并行读取的方式也很受欢迎。

---

更新：

在获得所有帮助后，我可以管理一些工作代码，谢谢。我要做一个简短的摘要。澄清一下：

1. 这是一种解析器，它有 3000 多行正则表达式和 自动生成的条件。

2. 我用于测试的输入是一个 POS 标记的文本，这个文件有 1071406 行。

3. 我的硬件是：SSD 磁盘、中档 i5 上一代和 8gb RAM DDR4。

结论：

1. 因为 cmets 建议 IO 操作使我的脚本变慢。
2. 所有建议都带来了改进，特别是包括处理成串行而不是逐行处理的建议。
3. 答案包含对未来工作非常有用的线程实现。
4. Framework Parallel::ForkManager 在执行时间上引入了很多延迟。我总是在 5 分钟后终止脚本，因为没有并行性的脚本大约需要 6 分钟。
5. Framework Parallel::Loops 引入了一些改进。该脚本大约需要 3 分钟才能完成。
6. 使用 GNU 并行是一种简单的优化方法。
7. 使用 Threads 包我得到了最好的时间，1 分 45 秒，但它非常接近 GNU 并行，所以你可以尝试一下，并努力移植代码。
8. 使用@ikegami 答案中的线程包读取一堆行，时间与应用@tanktalus 解决方案的时间相同，逐行读取。

最后，我将使用@ikegami 解决方案，我认为当数据量增加时会更好。例如，我将要处理的行数调整为 100.000，因为它比 10.000 获得更好的结果。这种差异大约是 8 秒。

下一个自然步骤是将所有内容写入文件而不是使用 STDOUT，我希望这有助于进一步减少时间。

Answer 1

$_ 从未设置，因为您从未分配给$_！

别忘了

while (<STDIN>) { ... }

简称

while (defined( $_ = <STDIN> )) { ... }

这意味着您希望使用以下内容：

$pl->while ( sub { defined( $_ = <STDIN> ) }, sub {
    print $_;
}

也就是说，破坏$_ 是个坏主意。它很可能被调用者中的 for (...) 别名为其他变量。

这意味着您应该使用以下内容：

my $line;
$pl->while ( sub { defined( $line = <STDIN> ) }, sub {
    print $line;
}

您可能会发现，将作品分解成更粗略的线条单位会产生更好的性能，因为它会降低偷听与作品的比率。

use constant WORK_UNIT_SIZE => 100;

my $done = 0;
my @lines;
$pl->while ( sub {
    @lines = ();
    return 0 if $done;

    while (@lines < WORK_UNIT_SIZE) {
        my $line = <>;
        if (!defined($line)) {
            $done = 1;
            return 0+@lines;
        }

        push @lines, $line;
    }

    return 1;
}, sub {
    for (@lines) {
        print $_;
    }
}

最后，您应该重用它们，而不是为每个工作单元创建一个新任务！下面使用线程演示了这一点。

use threads            qw( async );
use Thread::Queue 3.01 qw( );

use constant NUM_WORKERS    => 8;
use constant WORK_UNIT_SIZE => 100;

sub worker {
    my ($job) = @_;
    for (@$job) {
        print $_;
    }
}

my $q = Thread::Queue->new();
$q->limit(NUM_WORKERS * 4);

async { while (defined( my $job = $q->dequeue() )) { worker($job); } }
    for 1..NUM_WORKERS;

my $done = 0;    
while (!$done) {
    my @lines;
    while (@lines < WORK_UNIT_SIZE) {
        my $line = <>;
        if (!defined($line)) {
            $done = 1;
            last;
        }

        push @lines, $line;
    }

    $q->enqueue(\@lines) if @lines;
}

$q->end();
$_->join for threads->list;

Answer 2

我不知道使用Parallel::Loops 的具体好处（很可能有）。这与Parallel::ForkManager 相同，这是Parallel::Loops 使用的。

use warnings;
use strict;
use feature 'say';

use Parallel::ForkManager;   

my $max_procs = 30; 
my $pm = Parallel::ForkManager->new($max_procs);   

# Retrieve data returned by children in the callback
my %ret_data;      
$pm->run_on_finish( sub { 
    my ($pid, $exit, $ident, $signal, $core, $dataref) = @_; 
    $ret_data{$pid} = $dataref;
});

while (my $input = <STDIN>)
{
    chomp($input);

    $pm->start and next;
    my $ret = run_job($input);
    $pm->finish(0, \$ret);
}
$pm->wait_all_children;

foreach my $pid (keys %ret_data) {
    say "$pid returned: ${$ret_data{$pid}}";
}

sub run_job { 
    my ($input) = @_; 
    # your processing
    return $input;    # to have something to check
} 

此代码从子进程返回一个标量，一个值。您可以返回任何数据结构，请参阅文档中的 Retrieving data structures from child processes 和 this post 示例。

数据是通过文件返回的，对于大数据或许多快速进程，这可能会减慢速度。

如果在终端进行测试，则使用Ctrl-d 停止输入（或在chomp 之后添加last if $input !~ /\S/; 以以空行停止——但不是通过其他方式将数据传递给STDIN）。

---

已澄清每个STDIN 读取只是要处理的一行。然后我们应该在生成新进程之前收集更多行，否则开销太大。

my $num_lines_to_collect = 1000;

my @lines_to_process;         # collect lines for each fork

while (my $input = <STDIN>)
{
    chomp($input);
    push @lines_to_process, $input;
    next if $. % $num_lines_to_collect != 0;

    $pm->start and next;
    my $ret = run_job( \@lines_to_process );
    $pm->finish(0, \$ret);

    @lines_to_process = ();   # empty it for the next round
}
$pm->wait_all_children;

我们将行添加到数组@lines_to_process，并且仅当当前行号$. 是$num_lines_to_collect 的倍数时才继续触发新的fork。因此，每个$num_lines_collect 都会启动一个作业，因此每个作业都会处理这么多。我将它设置为1000，实验。

Answer 3

这里可能最简单的方法是创建一个线程池，每个线程都在同一个队列上进行侦听，然后让一个线程（可能是主线程）读取文件并将每一行推送到队列中。

use strict;
use warnings;
use Thread qw(async);
use Thread::Queue;

my $q = Thread::Queue->new();
$q->limit(32); # no point in reading in more than this into memory.

my @thr = map {
    async {
        while (defined (my $line = $q->dequeue()) ) {
            print $line;
        }
    };
} 1..4; # 4 worker threads

while (<STDIN>)
{
    $q->enqueue($_);
}
$q->end();

$_->join for Thread->list;

作为一个警告点，如果您需要将数据从工作线程推送回主线程，请注意。它不像其他语言那么简单。

更新：从线程切换到线程。虽然 async 函数被记录为返回线程对象，但这似乎对我不起作用，因此也必须更改连接。