如何在Perl中实现信号量线程通信？

Question

我的Perl脚本需要同时运行多个线程...

use threads ('yield', 'exit' => 'threads_only');
use threads::shared;
use strict;
use warnings;
 no warnings 'threads';
use LWP::UserAgent;
use HTTP::Request;
use HTTP::Async;
use ...

......并且这样的线程需要从web获取一些信息,因此HTTP::Async被使用.

my $request = HTTP::Request->new;
   $request->protocol('HTTP/1.1');
   $request->method('GET');
   $request->header('User-Agent' => '...');

my $async = HTTP::Async->new( slots            => 100,
                              timeout          => REQUEST_TIMEOUT,
                              max_request_time => REQUEST_TIMEOUT );

但是有些线程只有在其他线程这样说时才需要访问web.

my $start = [Time::HiRes::gettimeofday()];
my @threads = ();
foreach ... {
  $thread = threads->create(
    sub {
           local $SIG{KILL} = sub { threads->exit };
           my $url = shift;
           if ($url ... ) {
             # wait for "go" signal from other threads
           }
           my ($response, $data);
           $request->url($url);
           $data = '';
           $async->add($request);
           while ($response = $async->wait_for_next_response) {
             threads->yield();
             $data .= $response->as_string;
           }
           if ($data ... ) {
             # send "go" signal to waiting threads
           }
         }
       }, $_);

  if (defined $thread) {
    $thread->detach;
    push (@threads, $thread);
  }
}

可能有一个或多个线程在等待 "go"信号,并且可能存在一个或多个线程,这样的"go"信号可以发送.在一开始,信号量的状态是" 等待 ",一旦转向" 走 ",它就会保持这种状态.

最后,app会检查最长运行时间.如果线程运行时间过长,则会发送自终止信号.

my $running;
do {
  $running = 0;
  foreach my $thread (@threads) {
    $running++ if $thread->is_running();
  }
  threads->yield();
} until (($running == 0) || 
         (Time::HiRes::tv_interval($start) > MAX_RUN_TIME));
$running = 0;
foreach my $thread (@threads) {
  if ($thread->is_running()) {
    $thread->kill('KILL');
    $running++;
  }
}
threads->yield();

---

现在到了这一点.我的问题是:

1. 如何在脚本中最有效地编写等待"信号量"的代码(请参阅上面脚本中的注释).我应该只使用共享变量和一些虚拟 sleep 循环吗？

2. 我是否需要 在应用程序结束时添加一些sleep 循环,以便为线程提供自我毁灭的时间？

Answer 1

您可以查看Thread::Queue来执行这项工作。您可以设置一个队列来处理等待“go”信号的线程和发送“go”信号的线程之间的信号发送。这是我尚未测试的快速模型：

...
use Thread::Queue;
...
# In main body
my $q = Thread::Queue->new();
...
$thread = threads->create(
    sub {
           local $SIG{KILL} = sub { threads->exit };
           my $url = shift;
           if ($url ... ) {
             # wait for "go" signal from other threads
             my $mesg = $q->dequeue();
             # you could put in some termination code if the $mesg isn't 'go'
             if ($mesg ne 'go') { ... }
           }
           ...
           if ($data ... ) {
             # send "go" signal to waiting threads
             $q->enqueue('go');
           }
         }
       }, $_);
...

需要等待“go”信号的线程将等待出队方法，直到有东西进入队列。一旦消息进入队列，只有一个线程会抓取该消息并对其进行处理。

如果您希望停止线程以使它们不再运行，可以将停止消息插入到队列的头部。

$q->insert(0, 'stop') foreach (@threads);

Thread::Queue 和线程CPAN 发行版中的示例更详细地展示了这一点。

不幸的是，对于你的第二个问题，答案是，这取决于情况。当您继续终止线程时，需要进行什么样的清理才能干净关闭？如果地毯被从线下拉出来，最坏的情况会是什么？您需要随时计划进行清理工作。您可以做的另一个选择是等待每个线程实际完成。

我的评论询问是否可以删除调用的原因detach是因为此方法允许主线程退出而不关心任何子线程发生了什么。相反，如果您删除此调用，并添加：

$_->join() foreach threads->list();

到主块的末尾，这将要求主应用程序等待每个线程实际完成。

如果您保留该detach方法，那么如果您需要线程执行任何类型的清理，则需要在代码末尾休眠。当你调用detach一个线程时，你告诉 Perl 的是，当你的主线程退出时，你不关心该线程在做什么。如果主线程退出并且存在仍在运行且已分离的线程，则程序将结束且不会发出任何警告。但是，如果您不需要任何清理，并且仍然调用detach，请随时退出。