multithreading - 如何在等待先前线程完成时暂停Thread::Queue访问？

Question

所以这就是我现在所拥有的:

for my $action (@actionList){
  $q->enqueue([$_, $action]) for @component_dirs;

  print "\nWaiting for prior actions to finish up...\n";
  until (!defined($q->peek())) {}
}

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

但这似乎不起作用..是否有更好的方法强制队列等待先前的 $action项目完成，然后再次允许访问？

编辑:奇怪的是，它神奇地开始工作了……也许一直以来都在工作，而我只是没有使输出足够明显。无论哪种方式，我的问题仍然存在-有更好的方法吗？

Answer 1

您的代码不会等到上一个操作完成后才开始执行，它只会浪费CPU的资源，直到另一个线程开始处理最后一个作业为止。

对于“标志”之类的东西，通常应该使用信号量。信号量是带有up和down方法的线程安全计数器。例如，我们可以将信号与作业一起传递，该信号以计数零开始。每个线程在完成工作时都会增加信号量。我们的主线程尝试通过作业数来减少信号量，该信号量将阻塞直到所有线程都完成为止:

my $q = Thread::Queue->new;
my @workers = map { threads->create(\&worker, $q) } 1 .. $NUM_WORKERS;

for my $action (@actionList) {
    my $sem = Thread::Semaphore->new(0);
    $q->enqueue([$_, $action, $sem]) for @component_dirs;
    $sem->down(0+@component_dirs);  # wait for the threads
}

$q->end;
$_->join for @workers;

sub worker {
    my ($q) = @_;
    while (my $job = $q->dequeue) {
        my ($component, $action, $sem) = @$job;
        ...
        $sem->up;
    }
}

实际上，我们可以重用信号量。

有关更多详细信息，请参见 Thread::Semaphore docs。

此用法类似于 barriers。