多線程Perl腳本和crontab/init腳本

Question

我有一個使用線程的Perl腳本的問題。

它工作正常，當我手動啓動它，但是當我啓動使用crontab你我有這樣的反饋：

Perl的退出，活動的線程：

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

的PATH變量和SHELL變量是正確的crontad。

我儘量讓初始化腳本（啓動服務）和同樣的錯誤：

Feb 24 08:04:48 SERVER kernel: perl[103293]: segfault at 4a8 ip 00007f6cfd075dd9 sp 00007fffb93437c0 error 4 in libperl.so[7f6cfcfdf000+183000] Feb 24 08:04:49 SERVER test_ping[102238]: Perl exited with active threads: Feb 24 08:04:49 SERVER test_ping[102238]: 0 running and unjoined Feb 24 08:04:49 SERVER test_ping[102238]: 1 finished and unjoined Feb 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 變量都ok
3. 通過初始化腳本，不起作用
4. 嘗試修改perl腳本以加入所有線程，但腳本 在此之後被凍結。

任何人有一個建議？還有其他的東西要檢查/做什麼？

THK

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

下面的過程fonction：

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

---

下面是你得到你所應用上述修復後。我也移動了所有與線程有關的代碼，因爲它不屬於那裏。最後，我通過將輸出代碼移動到其自己的線程中，取消了對$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();