如何在 Perl 中开始多线程

Question

我有一个运行时间超过 13 小时的 perl 程序。我认为它可以从引入多线程中受益，但我以前从未这样做过，我不知道如何开始。

这是我的情况： 我有一个包含数百个文本文件的目录。我使用基本的 for 循环遍历目录中的每个文件并进行一些处理（对文件本身进行文本处理，在文件上调用外部程序并对其进行压缩）。完成后，我转到下一个文件。我继续以这种方式以串行方式一个接一个地处理每个文件。这些文件彼此完全独立，并且进程不返回任何值（除了成功/失败代码），因此这似乎是多线程的一个很好的候选者。

我的问题：

1. 如何重写我的基本循环以利用线程？那里似乎有几种线程模式。
2. 如何控制当前运行的线程数？如果我有 N 个可用内核，如何将线程数限制为 N 或 N - n？
3. 我需要手动管理线程数还是 Perl 会为我做这件事？

任何建议将不胜感激。

Answer 1

由于您的线程只是要启动一个进程并等待它结束，因此最好绕过中间人而只使用进程。除非您使用的是 Windows 系统，否则我建议您使用 Parallel::ForkManager。

use Parallel::ForkManager qw( );

use constant MAX_PROCESSES => ...;

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

my @qfns = ...;

for my $qfn (@qfns) {
   my $pid = $pm->start and next;
   exec("extprog", $qfn)
      or die $!;
}

$pm->wait_all_children();

如果您希望避免在 Windows 中使用不必要的中间线程，则必须使用类似于以下内容的东西：

use constant MAX_PROCESSES => ...;

my @qfns = ...;

my %children;
for my $qfn (@qfns) {
   while (keys(%children) >= MAX_PROCESSES) {
      my $pid = wait();
      delete $children{$pid};
   }

   my $pid = system(1, "extprog", $qfn);
   ++$children{$pid};
}

while (keys(%children)) {
   my $pid = wait();
   delete $children{$pid};
}

Answer 2

有人给了你一个分叉的例子。 Forks 在 Windows 上不是原生的，所以我倾向于使用线程。

为了完整起见 - 这里是线程如何工作的粗略概念（IMO 是更好的方法之一，而不是重生线程）。

#!/usr/bin/perl

use strict;
use warnings;

use threads;

use Thread::Queue;

my $nthreads = 5;

my $process_q = Thread::Queue->new();
my $failed_q  = Thread::Queue->new();

#this is a subroutine, but that runs 'as a thread'.
#when it starts, it inherits the program state 'as is'. E.g.
#the variable declarations above all apply - but changes to
#values within the program are 'thread local' unless the
#variable is defined as 'shared'.
#Behind the scenes - Thread::Queue are 'shared' arrays.

sub worker {
    #NB - this will sit a loop indefinitely, until you close the queue.
    #using $process_q -> end
    #we do this once we've queued all the things we want to process
    #and the sub completes and exits neatly.
    #however if you _don't_ end it, this will sit waiting forever.
    while ( my $server = $process_q->dequeue() ) {
        chomp($server);
        print threads->self()->tid() . ": pinging $server\n";
        my $result = `/bin/ping -c 1 $server`;
        if ($?) { $failed_q->enqueue($server) }
        print $result;
    }
}

#insert tasks into thread queue.
open( my $input_fh, "<", "server_list" ) or die $!;
$process_q->enqueue(<$input_fh>);
close($input_fh);

#we 'end' process_q  - when we do, no more items may be inserted,
#and 'dequeue' returns 'undefined' when the queue is emptied.
#this means our worker threads (in their 'while' loop) will then exit.
$process_q->end();

#start some threads
for ( 1 .. $nthreads ) {
    threads->create( \&worker );
}

#Wait for threads to all finish processing.
foreach my $thr ( threads->list() ) {
    $thr->join();
}

#collate results. ('synchronise' operation)
while ( my $server = $failed_q->dequeue_nb() ) {
    print "$server failed to ping\n";
}

如果您需要移动复杂的数据结构，我建议您查看Storable - 特别是freeze 和thaw。这些将让您在队列中轻松地在对象、散列、数组等周围进行洗牌。

请注意 - 对于任何并行处理选项，您都可以获得良好的 CPU 利用率，但不会获得更多的磁盘 IO - 这通常是一个限制因素。