Perl 多线程 - Perl 以活动线程退出:已完成和未连接

Question

我正在尝试在我的程序中创建一个子例程的线程（称为重组）。我使用了下面的代码来创建线程，我改编自这些线程 http://chicken.genouest.org/perl/multi-threading-with-perl/我使用了这段代码，因为这些线程是在循环中创建的，线程数取决于变量$ParentTally每个循环都不同（$ParentTally最多可以达到 1000 个，我不想一次运行 1000 个线程）

my $nb_process = 20;
my $nb_compute = $ParentTally;
my $i=0;
my @running = ();
my @Threads;
my %NewPopulation;
while (scalar @Threads < $nb_compute) {
    @running = threads->list(threads::running);
    if (scalar @running < $nb_process) {
        my $Offspring= threads->new (sub {Recombination(%Parent1Chromosome, %Parent2Chromosome)});
        push (@Threads, $Offspring);
        my $tid = $Offspring->tid;
    }
    @running = threads->list(threads::running);
    foreach my $thr (@Threads) {
        if ($thr->is_running()) {
           my $tid = $thr->tid;
        }
       elsif ($thr->is_joinable()) {
          my $tid = $thr->tid;
          my $Offspring1=$thr->join();
          $NewPopulation{$Offspring1}{'Tally'}+=1;
       }
    }
    @running = threads->list(threads::running);
    $i++;
}
while (scalar @running != 0) {
     foreach my $thr (@Threads) {
      if ($thr->is_joinable()){
         my $Offspring1=$thr->join(); 
         $NewPopulation{$Offspring1}{'Tally'}+=1;
      }
    }
    @running = threads->list(threads::running);
}

（注意：$ParentTally取自代码前面的另一个哈希，我的$ParentTally=$hashref->{'Tally'};所以程序的这一部分每次都会循环使用不同的$ParentTally值。%Parent1Chromosome和%Parent2Chromosome是在程序的早期创建的。子例程"重组"很长，所以我还没有发布它，但它返回一个整数。)

通常在运行程序

时（尽管并非总是如此，许多早期的代码都依赖于随机变量，因此程序永远不会运行相同）一旦完成，我就会得到"Perl退出活动线程："数字"完成和未连接"（"数字"因运行而异）。我以为：

 while (scalar @running != 0) {
     foreach my $thr (@Threads) {
      if ($thr->is_joinable()){
         my $Offspring1=$thr->join(); 
         $NewPopulation{$Offspring1}{'Tally'}+=1;
      }
    }

是否意味着所有线程都将在进入下一段代码之前完成？我做错了什么？（我以前从未使用过线程）。我曾研究过使用 http://www.perlmonks.org/?node_id=735931 但我并不真正了解如何使用 Thread：：Queue，并且找不到教程（并且不理解 http://perldoc.perl.org/Thread/Queue.html）。谢谢

Answer 1

不是对您的代码的修复，但这是我如何使用队列执行此操作的大纲（显然需要一些填充以满足您的目的）。 如果内存使用和问题，有很多方法可以改进这一点 - 生成的每个线程都需要所有作用域变量的完整副本;使用线程时很容易遇到内存问题

#!/usr/bin/perl
use strict ;
use threads ;
use Thread::Queue ;
my $threadCount = 2 ;
my $DataQueue = Thread::Queue->new() ;
my $ReportQueue = Thread::Queue->new() ;
my $threads = [] ;
# create pool of worker threads
for ( my $i = 0 ; $i<$threadCount ; $i ++ ){
    push( @$threads, threads->create( &doStuff, $DataQueue, $ReportQueue ) ) ;
}
# array of data on which the threads have to work
my @array ;
# put work onto queue for threads to process
foreach my $workItem ( @array ){
   $DataQueue->enqueue( $workItem );
}
# enqueue undef for each worker to tell it no more work
# then wait for them all to join
$DataQueue->enqueue( (undef) x $threadCount ) ;
$_->join for @$threads ;
my %NewPopulation ;
# read the output of the threads from ReportQueue
while ( my $reportItem = $ReportQueue->dequeue() ){
    $NewPopulation{$reportItem}{'Tally'}++ ;
}
# display tallys
for my $offspring ( keys %NewPopulation ){
    print "Offspring $offspring Tally => " . $NewPopulation{$offspring}{'Tally'} . "n" ;
}
sub doStuff{
    my ( $DataQueue, $ReportQueue ) = @_ ;
    while ( my $inputHash = $DataQueue->dequeue() ){
        my $result ;
        # do things here - the logic in your Recombination sub
        # return result to report queue
        $ReportQueue->enqueue($result) ;
    }
   # Enqueue undef to report queue so report thread knows we're done
   $ReportQueue->enqueue( undef ) ;
}

Answer 2

我相信

错误在最后一个while循环中：

while (scalar @running != 0) {
     foreach my $thr (@Threads) {
      if ($thr->is_joinable()){
         my $Offspring1=$thr->join(); 
         $NewPopulation{$Offspring1}{'Tally'}+=1;
      }
    }
    @running = threads->list(threads::running);
}

根据线程文档，仅当线程已完成运行、尚未分离且尚未加入时，对 is_joinable 的调用才会返回 true。 我的猜测是，当您进入本节时，您仍然有正在运行的线程，因此您跳过它们。 您可以像在上一个while循环中所做的那样进行另一次调用，以is_running以查看是否仍有线程正在运行并以某种方式处理线程。