multithreading - 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」シグナルを待機している1 つ以上のスレッドが存在する可能性があり、そのような「go」シグナルが送信できるスレッドが1 つ以上存在する可能性があります。セマフォの状態は最初は「待ち」で、一旦「進行」になるとそのままです。

最後に、アプリは最大実行時間をチェックします。スレッドの実行時間が長すぎる場合、自己終了シグナルが送信されます。

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();

さて、要点です。私の質問は次のとおりです。

スクリプトで待機中の「セマフォ」を最も効果的にコーディングするにはどうすればよいですか (上記のスクリプトのコメントを参照)。 ダミーループで共有変数だけを使用する必要がありますか?sleep
自己破壊のためにスレッドに時間を与えるために、アプリの最後にループを追加する必要がありますか?sleep

score 3 · Accepted Answer

この作業を実行するには、 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」シグナルを待機する必要があるスレッドは、何かがキューに入るまで dequeue メソッドで待機します。メッセージがキューに入ると、1 つのスレッドと 1 つのスレッドだけがメッセージを取得して処理します。

スレッドが実行されないように停止したい場合は、停止メッセージをキューの先頭に挿入できます。

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

Thread::Queue とスレッドの CPAN ディストリビューションには、これをより詳細に示す例があります。

2 番目の質問への回答ですが、残念ながら状況によって異なります。スレッドの終了に進む場合、クリーンシャットダウンにはどのような種類のクリーンアップが必要ですか? ラグが糸の下から引っ張られた場合に起こりうる最悪のシナリオは何ですか? いつでもクリーンアップが発生するように計画する必要があります。あなたができる他のオプションは、各スレッドが実際に完了するのを待つことです。

呼び出しを削除できるかどうかを尋ねる私のコメントの理由detachは、このメソッドにより、メインスレッドが終了し、子スレッドに何が起こっていたかを気にしないためです。代わりに、この呼び出しを削除して以下を追加すると:

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

メインブロックの最後まで、これにより、メインアプリケーションは各スレッドが実際に完了するまで待機する必要があります。

メソッドをそのままにしておくdetachと、スレッドで何らかのクリーンアップを実行する必要がある場合、コードの最後でスリープする必要があります。スレッドを呼び出すときdetach、Perl に伝えていることは、メインスレッドが終了したときにスレッドが何をしているかは気にしないということです。メインスレッドが終了し、切断されたスレッドがまだ実行されている場合、プログラムは警告なしで終了します。ただし、クリーンアップを必要とせず、それでもを呼び出す場合はdetach、いつでも好きなときに自由に終了してください。

score -1 · Accepted Answer

このようなものを試してみてください....

#!/usr/bin/perl

use threads;
use threads::shared;

$|=1;

my ($global):shared;
my (@threads);

push(@threads, threads->new(\&mySub,1));
push(@threads, threads->new(\&mySub,2));
push(@threads, threads->new(\&mySub,3));

$i = 0;

foreach my $myThread(@threads)

{
    my @ReturnData = $myTread->join ;
    print "Thread $i returned: @ReturnData\n";
    $i++;
}

sub mySub
{
    my ($threadID) = @_;

    for(0..1000)
    {
        $global++;
        print "Thread ID: $threadID >> $_ >> GLB: $global\n";
        sleep(1);
    }   
    return( $id );
}

multithreading - Perlでセマフォスレッド通信を実装するには?

2 に答える 2

Related

Reference