使用 Perl ithread 的奇怪变量行为

Question

我正在尝试基于稍微改变的老板/工人模型来实现多线程应用程序。基本上，主线程创建了几个老板线程，每个老板线程又产生两个工作线程（可能更多）。这是因为老板线程每个处理一个主机或网络设备，工作线程可能需要一段时间才能完成工作。

我正在使用Thread::Pool 来实现这个概念，到目前为止效果很好；我也不认为我的问题与Thread::Pool 有关（见下文）。前面非常简化的伪代码：

use strict;
use warnings;

my $bosspool = create_bosspool();   # spawns all boss threads
my $taskpool = undef;               # created in each boss thread at
                                    # creation of each boss thread 

# give device jobs to boss threads
while (1) {
  foreach my $device ( @devices ) {
    $bosspool->job($device);
  }

  sleep(1);
}

# This sub is called for jobs passed to the $bosspool
sub process_boss
{
  my $device = shift;

  foreach my $task ( $device->{tasks} ) {
    # process results as they become available
    process_result() while ( $taskpool->results );
    # give task jobs to task threads
    scalar $taskpool->job($device, $task);
    sleep(1); ### HACK ###
  }

  # process remaining results / wait for all tasks to finish
  process_result() while ( $taskpool->results || $taskpool->todo );

  # happy result processing
}

sub process_result
{
  my $result = $taskpool->result_any();

  # mangle $result
}

# This sub is called for jobs passed to the $taskpool of each boss thread
sub process_task
{
  # not so important stuff

  return $result;
}

顺便说一句，我不使用monitor() 例程的原因是因为我必须等待$taskpool 中的所有作业完成。现在，这段代码工作得非常好，除非你删除### HACK ### 行。如果您不睡觉，$taskpool->todo() 将无法提供正确数量的仍处于打开状态的工作，如果您添加它们或接收它们的结果太“快”。就像，您总共添加了 4 个工作，但 $taskpool->todo() 之后只会返回 2 个（没有待处理的结果）。这会产生各种有趣的效果。

好的，所以Thread::Pool->todo() 是废话，让我们尝试一种解决方法：

sub process_boss
{
  my $device = shift;

  my $todo = 0;

  foreach my $task ( $device->{tasks} ) {
    # process results as they become available
    while ( $taskpool->results ) {
      process_result();
      $todo--;
    }
    # give task jobs to task threads
    scalar $taskpool->job($device, $task);
    $todo++;
  }

  # process remaining results / wait for all tasks to finish
  while ( $todo ) {
    process_result();
    sleep(1); ### HACK ###
    $todo--;
  }
}

只要我保留### HACK ### 行，这也可以正常工作。如果没有这一行，这段代码将重现Thread::Pool->todo() 的问题，因为$todo 不仅会减少1，还会减少2 甚至更多。

我只用一个老板线程测试了这段代码，所以基本上没有涉及多线程（当涉及到这个子例程时）。 $bosspool、$taskpool 尤其是 $todo 不是 :shared，不可能有副作用，对吧？这个子程序中发生了什么，它只由一个老板线程执行，没有共享变量、信号量等？

Answer 1

我建议实现“工作”线程模型的最佳方式是使用Thread::Queue。做这样的事情的问题是确定队列何时完成，或者项目是否已出列并等待处理。

使用Thread::Queue，您可以使用while 循环从队列中获取元素，并使用end 从队列中获取元素，这样while 循环返回undef 并且线程退出。

所以您并不总是需要多个“老板”线程，您可以使用多种不同风格的worker 和输入队列。我会质疑为什么在那种情况下你需要一个“老板”线程模型。似乎没有必要。

参考： Perl daemonize with child daemons

#!/usr/bin/perl

use strict;
use warnings;
use threads;
use Thread::Queue;

my $nthreads = 4;

my @targets = qw ( device1 device2 device3 device4 );

my $task_one_q = Thread::Queue->new();
my $task_two_q = Thread::Queue->new();

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

sub task_one_worker {
    while ( my $item = task_one_q->dequeue ) {

        #do something with $item

        $results_q->enqueue("$item task_one complete");
    }
}

sub task_two_worker {
    while ( my $item = task_two_q->dequeue ) {

        #do something with $item

        $results_q->enqueue("$item task_two complete");
    }
}

#start threads;

for ( 1 .. $nthreads ) {
    threads->create( \&task_one_worker );
    threads->create( \&task_two_worker );
}

foreach my $target (@targets) {
    $task_one_q->enqueue($target);
    $task_two_q->enqueue($target);
}

$task_one_q->end;
$task_two_q->end;

#Wait for threads to exit.

foreach my $thr ( threads->list() ) {
    threads->join();
}

$results_q->end();

while ( my $item = $results_q->dequeue() ) {
    print $item, "\n";
}

如果您愿意，您可以使用boss 线程执行类似的操作 - 您可以为每个boss 创建一个队列，并通过引用将其传递给工作人员。我不确定这是否有必要。