multithreading - 在 Perl 线程中调用 system() 时产生僵尸进程

Question

在我的测试环境中有两个Linux节点(hostA和hostB)，我需要触发一个脚本(worker.sh)在所有节点上同时运行，worker.sh已经放在所有节点中，所以我使用我的 Perl 脚本 (master.pl) 中的线程模块，这里是代码片段:

use threads(stringify);

sub runByThreads{
   my($count,$funcion,$host_ref,$cmd) = @_;
   @hostlist = @{$host_ref};

   my $thread;
   my @failNodes;

   for (my $i=0;$i<$count;$i++) {
      my $host =@hostlist[$i];
      $thread = threads->create($funcion,$host,$cmd);
      $parserState{$thread} = $host;
      $thread_num ++;
   }

   while ($thread_num != 0) {      # stuck in this while loop
      foreach my $subthread(threads->list(threads::joinable)) {
         my $ret = $subthread->join();
         if ($ret != 0) {
            ....
         }
         $thread_num --;
     }
     sleep 2;
   }
}

sub runCmd {
    my ($host,$cmd) = @_;

    chomp($localhost = `hostname -f`);
    if ($localhost eq $host) {
        $ret = system("source /etc/profile; $cmd");
    } else {
        $ret = system("ssh -o StrictHostKeyChecking=no ".$host." \"source /etc/profile; ". $cmd."\"");
    }
    return $ret;
}


main {
    my @servers = qw/hostA hostB/
    my $nodecount = scalar(@servers);
    my $arg = "--node";

    $cmd = "$HOME/worker.sh "."$arg";
    my @ret = &runByThreads($nodecount,\&runCmd,\@servers,$cmd);
    if ( scalar(@ret) != 0) {
        $failNum += 1;
    }
}

&main;

此perl脚本在hostA上运行，正常情况下ps命令显示:

0 S optitest  9338  9337  0  80   0 - 50630 pipe_w 06:57 ?        00:00:00 /usr/bin/perl master.pl
0 S optitest  9992  9338  0  80   0 - 26536 wait   06:57 ?        00:00:00 sh -c source /etc/profile; /home/jack/linux/worker.sh --node
0 S optitest 10023  9338  0  80   0 - 14151 poll_s 06:57 ?        00:00:00 ssh -o StrictHostKeyChecking=no hostB source /etc/profile; /home/jack/linux/worker.sh --node
0 S optitest 10757 10741  0  80   0 -  1608 pipe_w 06:59 ?        00:00:00 grep 9338

但是有时候ps会显示存在一个defunct进程，这个defunct进程会导致master.pl卡在while循环中，

0 S optitest  6503  6502  1  80   0 - 50628 pipe_w 05:51 ?        00:00:00 /usr/bin/perl master.pl
0 Z optitest  7496  6503  0  80   0 -     0 exit   05:51 ?        00:00:00 [hostname] <defunct>
0 S optitest  7497  6503  0  80   0 - 26536 wait   05:51 ?        00:00:00 sh -c source /etc/profile; cd /home/jack/linux/worker.sh --node

我知道僵尸进程是一个已经完成执行(通过退出系统调用)但在进程表中仍有一个条目的进程，这发生在子进程中，仍然需要该条目以允许父进程读取其子进程的退出状态:一旦通过 wait 系统调用读取了退出状态，僵尸的条目就会从进程表中删除，并被称为“收割”

我很困惑在我的测试中defunct进程是如何生成的，defunct进程应该是通过ssh在hostB上运行work.pl的那个进程，但我发现它似乎在Perl系统创建时立即成为defunct进程调用，因为我没有看到它运行的任何输出，甚至没有执行 worker.sh 第一行中的 'echo'。

还有一点很奇怪，在worker.sh中，调用了一些脚本在后台运行，如果我在hostB上清空worker.sh，也会出现defunct问题，但是如果我在hostB上清空worker.sh hostA 和 hostB，我再也没有看到过时的问题。

很抱歉发了这么长的帖子，我正在尽力让我的问题更清楚，你能帮我检查一下哪里出了问题吗，我在使用线程模块时是否遗漏了什么，或者有一些问题threads 模块，因为我注意到官方不鼓励在 perl 中使用基于解释器的线程。 http://perldoc.perl.org/threads.html

Answer 1

线程在 perldoc 中被列为“discouraged”。就个人而言，我发现它们工作得很好，它们只是有点违反直觉 - 它们不是可能假设的轻量级结构(基于其他线程模型)。

我会注意到 - 自收割僵尸的通用解决方案是设置 $SIG{'CHLD'} 例如:http://perldoc.perl.org/perlipc.html但如果您要捕获返回码，那可能不是一个好主意。不过，您或许可以执行 open 和 waitpid。

所以我通常不会建议使用它们，除非您有需要进行大量线程间通信的场景。 Parallel::ForkManager 通常效率更高。

如果您确实必须使用它们 - 我不会做您正在做的事情，也不会为每个“作业”生成一个线程，而是使用带有 Thread::Queue 的工作线程模型。

我不能肯定地说，但我怀疑你的问题之一是这一行:

$cmd = "$HOME/worker.sh "."$arg";

因为 perl 将插入 $HOME - 而您没有定义它，因此它是空的。

您真的应该打开 strict 和 warnings 并因此清除任何错误 - 您的代码有很多。

但话虽如此 - 除非我遗漏了一些东西，否则你的代码比它需要的要复杂得多 - 看起来你在这里所做的只是运行并行的 ssh 命令。

所以我建议你最好是这样的:

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

use threads;
use Thread::Queue;

my @servers = qw/hostA hostB/;

my $cmd         = '$HOME/worker.sh --node';
my $threadcount = 2;

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

sub worker {
    while ( my $hostname = $hostq->dequeue ) {
        my $output =
            qx( ssh -o StrictHostKeyChecking=no $hostname \"source /etc/profile; $cmd\" );
        if ($?) {
            $errorq->enqueue("$hostname: $output");
        }
    }
}


$hostq->enqueue(@servers);
for ( 1 .. $threadcount ) {
    my $thr = threads->create( \&worker );
}
$hostq->end();

foreach my $thr ( threads->list ) {
    $thr->join;
}
$errorq->end();
while ( my $error = $errorq->dequeue ) {
    print "ERROR: $error\n";
}

或者，使用 Parallel::ForkManager:

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

my @servers = qw/hostA hostB/;

my $cmd     = '$HOME/worker.sh --node';
my $manager = Parallel::ForkManager->new(5);    #fork limit.

foreach my $hostname (@servers) {
    $manager->start and next;
    my $output =
        qx( ssh -o StrictHostKeyChecking=no $hostname \"source /etc/profile; $cmd\" );
    if ($?) {
        print "ERROR: $hostname $output\n";
    }
    $manager->finish;
}

$manager->wait_all_children();