多线程 Perl 脚本和 crontab/init 脚本

Question

我对使用线程的Perl 脚本有疑问。

当我手动启动它时它工作正常，但是当我使用 crontab 启动它时，我有这个反馈：

Perl 以活动线程退出：

    0 running and unjoined
    1 finished and unjoined
    0 running and detached

PATH 变量和 SHELL 变量在 crontad 上是正确的。

我尝试制作一个初始化脚本（作为服务启动）并出现同样的错误：

2 月 24 日 08:04:48 服务器内核：perl[103293]：4a8 ip 的段错误 00007f6cfd075dd9 sp 00007fffb93437c0 错误 4 英寸 libperl.so[7f6cfcfdf000+183000] 2 月 24 日 08:04:49 服务器 test_ping [102238]：Perl 以活动线程退出：2 月 24 日 08:04:49 服务器 test_ping [102238]：0 运行和未加入 2 月 24 日 08:04:49 服务器 test_ping [102238]：1 完成并未加入 2 月 24 日 08:04:49 SERVER test_ping[102238]: 0 running and detached

所以我也尝试过修改perl：

for my $thread (threads->list) {                                                                                                                
$thread->join();                                                                               
}

代替

for my $thread (threads->list) {                                                                                                                
$thread->detach();                                                                               
}

经过修改后，当我手动启动脚本时，这个脚本似乎被卡住/冻结了。

所以要恢复这就是我的全部检查：

1. 手动执行就可以了
2. 通过 crontab 不起作用，检查 PATH 变量和 SHELL 变量没问题
3. 通过初始化脚本，不起作用
4. 尝试修改 perl 脚本以加入除脚本之外的所有线程 之后就结冰了。

有人有建议吗？还有什么要检查/做的吗？

想

 use lib '/usr/local/perf/lib';
use lib '/usr/share/perl5';
use threads;
use Thread::Queue;
use SNMP::Persist qw(&define_oid &start_persister &define_subtree);
use Schedule::ByClock;
use Time::HiRes qw( usleep ualarm gettimeofday tv_interval );

use strict;
#use warnings;
use constant DEBUG => 0;
use constant DEBUG2 => 1;

if ($#ARGV + 1 != 2) {
 print "usage: test_ping.pl OUTPUTFILE INPUTFILE \n";
 exit;
}

my $output_file=$ARGV[0];
my $data_file=$ARGV[1];
shift @ARGV;
shift @ARGV;

#start the thread serving answers
start_persister();

#create queue for processing commands
my $q_queue = new Thread::Queue;
my $r_queue = new Thread::Queue;

#create threads for processing queues
for(my $i= 0; $i < $thread_num; $i++) {
        threads->create(\&process) -> detach();
}
        my $datestring=localtime();

        my %subtree;
        my @raw_data;

        my ($q_line, @q_split);
        my ($r_line, @r_split);
        my $index=0;

        # open file to get data
        open(DAT, $data_file) || die("Could not open file!");
        @raw_data=<DAT>;
        close(DAT);

        # enqueue all lines to be process by threads
        foreach $q_line (@raw_data) {
                chomp($q_line);
                $q_line =~ s/^\s+//;
                $q_line =~ s/\s+$//;
                next if ($q_line =~ /^#.*/);
                next if ($q_line eq "");
                next if ($q_line =~ /^\|/);

                @q_split=split(/\|/,$q_line);
                next if (!($q_split[0] eq "icmp" || $q_split[0] eq "tcp" || $q_split[0] eq "ldap" || $q_split[0] eq "dig" ));

                $q_queue->enqueue(++$index ."|". $q_line);
        }

        while ($index != 0 && ($r_line = $r_queue->dequeue)) {

                open(my $fh, '>>', $output_file) or die "Could not open file '$output_file' $!";
                print $fh $datestring."|";
                print $fh $r_line."\n";
                close $fh;
                @r_split=split(/\|/,$r_line);
                $index--;
        }

        for my $thread (threads->list) {                                                                                                                
            $thread->detach();                                                                               
    }  

下面的过程函数：

sub process {
    # my @hotefqdn = split(/\./, `hostname`);
    # my $hote=$hotefqdn[0];
    my ($q_line,@q_split,$q_index,$q_query);
    my ($q_module,$q_type,$q_name,$q_host,$q_port,$q_ssl,$q_send,$q_expect,$q_quit);
    my ($q_lookup,$q_record);
    my ($q_base_dn,$q_attr,$q_binddn,$q_password,$q_warn_time,$q_crit_time,$q_timeout);
    my ($r_tab);

    while ($q_line = $q_queue->dequeue) { 

            @q_split=split(/\|/,$q_line);

            $q_index=$q_split[0];
            $q_module=$q_split[1];

            if ($q_module eq "icmp") {
                    $q_type=$q_split[2];
                    $q_name=$q_split[3];
                    $q_host=$q_split[4];
                    $q_query="$q_host (ping)";
                    print "query=$q_query\n" if(DEBUG);
                    $r_tab=icmp_query($q_host);
            }
            elsif ($q_module eq "tcp") {
                    $q_type=$q_split[2];
                    $q_name=$q_split[3];
                    $q_query="$q_host ($q_type:$q_port)";
                    print "query=$q_query\n" if(DEBUG);
                    $r_tab=tcp_query($q_host,$q_port,$q_ssl,$q_send,$q_expect,$q_quit);
            }
            elsif ($q_module eq "ldap") {
                    $q_type=$q_split[2];
                    $q_name=$q_split[3];
                    print "query=$q_query\n" if(DEBUG);
                    $r_tab=ldap_query($q_host,$q_base_dn,$q_port,$q_attr,$q_binddn,$q_password,$q_warn_time,$q_crit_time,$q_timeout);
            }
            elsif ($q_module eq "dig") {
                    $q_type=$q_split[2];
                    $q_name=$q_split[3];
                    $q_query="$q_lookup($q_record) @".$q_host;
                    print "query=$q_query\n" if(DEBUG);
                    $r_tab=dig_query($q_host,$q_port,$q_lookup,$q_record,$q_expect);
            }

            $r_queue->enqueue($q_index."|".$q_name."|".$q_type."|".$q_query."|".$r_tab->{'min'}."|".$r_tab->{'med'}."|".$r_tab->{'avg'}."|".$r_tab->{'max'}."|".$r_tab->{'dev'}."|".$r_tab->{'loss'}."|".$r_tab->{'err'});
    }

}

Answer 1

首先，不要分离线程。当你这样做时，你不能等待他们完成。

for (my $i= 0; $i < $thread_num; $i++) {
    threads->create(\&process) -> detach();
}

...

for my $thread (threads->list) {
    $thread->detach();
}  

应该是

for (1..$thread_num) {
    threads->create(\&process);
}

...

... Tell the threads to finish up ...

for my $thread (threads->list) {
    $thread->join();
}  

---

现在的问题是：为什么你的线程没有完成？好吧，你永远不会告诉他们退出，所以他们永远不会这样做！您需要要求他们退出，这可以通过添加以下内容来实现：

$q_queue->end();

---

以下是您在应用上述修复后得到的结果。我还将所有与线程相关的代码移出process，因为它不属于那里。最后，我通过将输出代码移到自己的线程中，消除了对$index 的依赖。

sub process {
   my ($q_line) = @_;
   ...
   return join("|", $q_index, $q_name, $q_type, $q_query, @$r_tab{qw( min med avg max dev loss err )});
}

my $request_q  = Thread::Queue->new();
my $response_q = Thread::Queue->new();

my @worker_threads;
for (1..$thread_num) {
   push @worker_threads, async {
      while (defined( my $request = $request_q->dequeue() )) {
         $response_q->enqueue( process($request) );
      }
   };
}

my $output_thread = do {
   my $datestring = localtime();

   open(my $fh, '>', $output_file)
      or die("Can't create file \"$output_file\": $!\n");

   async {
      while (defined( my $response = $response_q->dequeue() )) {
         print($fh "$datestring|$response\n");
      }
   }
};

{    
   my %protos = map { $_ => 1 } qw( icmp tcp ldap dig );

   open(my $fh, '<', $data_file)
      or die("Can't open file \"$data_file\": $!\n");

   my $index = 0;
   while (<$fh>) {
      s/^\s+//;
      s/\s+\z//;
      next if $_ eq "" || /^#/;

      my ($proto) = split /\|/;
      next if !$protos{$proto};

      $request_q->enqueue(++$index ."|". $_);
   }
}

$request_q->end();
$_->join() for @worker_threads;

$response_q->end();
$output_threads->join();