複数の子プロセスをフォークして読み込む方法は？

Question

私の主な目標は、オブジェクトの大きなリストのためにいくつかの（外部的に時間がかかる）高価な作業を行うことです。そのために、もし私がそれをまっすぐにするなら、それは多くの時間がかかります。だから、私はパラレルモードに行き、いくつかの子プロセスをフォークする（4-8、みましょう）。メイン（親）プロセスでは、ワンプロセスバージョンの進捗状況に関する同じ統計情報をすべて出力したかったのです。

しかし、私は4つの子プロセスをフォークしていくつかの作業を行うと、それらが生きていることがわかりますが、実際に何かをして親に情報を戻しているだけです。

これまで私が行ったコードは次のとおりです。時間のかかる部分は、その動作を非常にうまくシミュレートするランダムなスリープ状態に惑わされています。

#!/usr/bin/env perl 
use strict; 
use warnings; 

use DateTime; 
use DateTime::Format::HTTP; 
use Time::HiRes; 

my @to_be_processed = (1..300000); 
my @queues; 
my $nprocs = 4; 

my $parent_from_child; 
my @child_from_parent; 
my @child_to_parent; 

$SIG{CHLD} = 'IGNORE'; 
$|=1; # autoflush 

my %stat = (
    total   => scalar(@to_be_processed), 
    processed  => 0, 
    time_started => [Time::HiRes::gettimeofday], 
); 

# divide the list into queues for each subprocess 
for (my $i = 0; $i < $stat{total}; $i++) { 
    my $queue = $i % $nprocs; 
    push @{$queues[$queue]}, $to_be_processed[$i]; 
} 

# for progress simulation 
srand (time^$$); 

for (my $proc = 0; $proc < $nprocs; $proc++) { 

    # set up the pipes 
    pipe $parent_from_child, $child_to_parent[$proc]  or die "pipe failed - $!"; 

    # fork 
    defined(my $pid = fork) or die "fork failed - $!"; 

    if ($pid) { 
     # parent 
     close $child_to_parent[$proc]; 
     printf("[%u] parent says: child %u created with pid %u\n", $$, $proc, $pid); 
    } 
    else { 
     # child 
     close $parent_from_child; 
     open(STDOUT, ">&=" . fileno($child_to_parent[$proc])) or die "open failed - $!"; 

     warn(sprintf("[%u] child alive with %u entries\n", $$, scalar(@{$queues[$proc]}))); 

     foreach my $id (@{$queues[$proc]}) { 
      printf("START: %s\n", $id); 

      # simulation of progress 
      my $random_microseconds = int(rand(3000000))+200000; 
      warn(sprintf("[%u] child 'works' for %u microseconds", $$, $random_microseconds)); 
      Time::HiRes::usleep($random_microseconds); 

      printf("DONE\n") 
     } 
     exit(0); 
    } 
} 

# parent: receive data from children and print overall statistics 
while (<$parent_from_child>) { 
    chomp(my $line = $_); 

    if ($line =~ m/^START: (\S+)/) { 
     my ($id) = @_; 

     printf("%6u/%6u", $stat{processed}, $stat{total}); 
     if ($stat{time_avg}) { 
      my $remaining = ($stat{total} - $stat{processed}) * $stat{time_avg}; 
      my $eta = DateTime->from_epoch(epoch => time + $remaining); 
      $eta->set_time_zone('Europe/Berlin'); 
      printf(" (ETA %s)", DateTime::Format::HTTP->format_isoz($eta)); 
     } 
     printf("\r"); 
    } 
    elsif ($line =~ /^DONE/) { 
     $stat{processed}++; 
     $stat{time_processed} = Time::HiRes::tv_interval($stat{time_started}); 
     $stat{time_avg}  = $stat{time_processed}/$stat{processed}; 
    } 
    else { 
     printf("%s\n", $line); 
    } 
} 

通常、警告は削除する必要があります。 これを実行すると、1つの子だけが動作することがわかります。 私の質問は：なぜですか？私のミスはどこにあり、どのようにして仕事をしているのですか？

おかげ K.

Answer 1

あなたはstraceの下のperlを実行することができ、あなたがあなたの子供の生活はかなり不足していることがわかりますし、次のようになります。

close(3)        = 0 
ioctl(4, SNDCTL_TMR_TIMEBASE or TCGETS, 0x7fff753b3a10) = -1 EINVAL (Invalid argument) 
lseek(4, 0, SEEK_CUR)     = -1 ESPIPE (Illegal seek) 
fstat(4, {st_mode=S_IFIFO|0600, st_size=0, ...}) = 0 
dup2(4, 1)        = 1 
dup(4)         = 3 
fcntl(4, F_GETFD)      = 0x1 (flags FD_CLOEXEC) 
dup2(3, 4)        = 4 
fcntl(4, F_SETFD, FD_CLOEXEC)   = 0 
close(3)        = 0 
fcntl(1, F_SETFD, 0)     = 0 
write(2, "[30629] child alive with 75000 e"..., 39) = 39 
brk(0x3582000)       = 0x3582000 
write(1, "START: 1\n", 9)    = -1 EPIPE (Broken pipe) 
--- SIGPIPE (Broken pipe) @ 0 (0) --- 

これが理由です：

pipe $parent_from_child, $child_to_parent[$proc]  or die "pipe failed - $!"; 

パイプに間違った引数で配列を使用しました。 の両サイドを親に開いた状態に保つ必要があります。代わりに、親が書き込み側をすべて開いたままにすることができるように配列を設定しました（ただし、親ブロックでは書き込み側をすぐに閉じます）。したがって、次回のループでは、pipeが新しいハンドルを作成し、$parent_from_childに割り当てます。したがって古い値には参照がなくなり、perlはそれをクリーンアップし、ファイルハンドルを閉じます。だからあなたの子供たちはSIGPIPEの最後の人を除いてすべて死ぬ。

私はあなたがそのハンドルを再利用して複数の書き込みハンドルを割り当てることができるという印象を受けていると思います。できません。 pipeは、常に新しい読み取りハンドルと新しい書き込みハンドルを作成します。

実際に同じ読み取りハンドルを共有したい場合（おそらくそうではありませんが、2つのクライアントからの出力がインターリーブされると破損することになります）、ループの外側で1回だけ作成してください。すべての子は同じ書き込みハンドルをfork経由で継承します。おそらく、あなたは子供1人あたり1つが必要です。利用可能な出力を確認するにはselectループを使用し、それらを読みとる必要があります。

また、CPANには既製のソリューション（または10個）が用意されています。