change time outs on selects

[spider.git] / perl / cluster.pl

diff --git a/perl/cluster.pl b/perl/cluster.pl
index ebe7ce0a51f9967ea56e58eeae5e9684fedeb986..ceb099ca7eb4f6b64977447e6a800d7c5be0c1e6 100755 (executable)
--- a/perl/cluster.pl
+++ b/perl/cluster.pl
@@ -61,13 +61,13 @@ use Filter;
 use Local;
 use Fcntl ':flock'; 
 
-use Carp;
+use Carp qw(cluck);
 
 package main;
 
 @inqueue = ();                                 # the main input queue, an array of hashes
 $systime = 0;                                  # the time now (in seconds)
-$version = "1.27";                             # the version no of the software
+$version = "1.33";                             # the version no of the software
 $starttime = 0;                 # the starting time of the cluster   
 $lockfn = "cluster.lock";       # lock file name
       
@@ -89,6 +89,8 @@ sub already_conn
        sleep(1);
        dbg('chan', "-> Z $call bye\n");
        $conn->send_now("Z$call|bye"); # this will cause 'client' to disconnect
+       sleep(1);
+       $conn->disconnect;
 }
 
 # handle incoming messages
@@ -105,9 +107,16 @@ sub rec
        # set up the basic channel info - this needs a bit more thought - there is duplication here
        if (!defined $dxchan) {
                my ($sort, $call, $line) = $msg =~ /^(\w)(\S+)\|(.*)$/;
+
+                # is there one already connected to me ? 
+               my $user = DXUser->get($call);
+               if (DXChannel->get($call)) {
+                       my $mess = DXM::msg($lang, $user->sort eq 'A' ? 'concluster' : 'conother', $call);
+                       already_conn($conn, $call, $mess);
+                       return;
+               }
                
                # is there one already connected elsewhere in the cluster (and not a cluster)
-               my $user = DXUser->get($call);
                if ($user) {
                        if (($user->sort eq 'A' || $call eq $myalias) && !DXCluster->get_exact($call)) {
                                ;
@@ -167,17 +176,43 @@ sub cease
                Local::finish();   # end local processing
        };
        dbg('local', "Local::finish error $@") if $@;
-       
+
+       # disconnect users
+       foreach $dxchan (DXChannel->get_all()) {
+               next if $dxchan->is_ak1a;
+               disconnect($dxchan) unless $dxchan == $DXProt::me;
+       }
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+
+       # disconnect nodes
        foreach $dxchan (DXChannel->get_all()) {
+               next unless $dxchan->is_ak1a;
                disconnect($dxchan) unless $dxchan == $DXProt::me;
        }
        Msg->event_loop(1, 0.05);
        Msg->event_loop(1, 0.05);
        Msg->event_loop(1, 0.05);
        Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
+       Msg->event_loop(1, 0.05);
        DXUser::finish();
+       dbg('chan', "DXSpider version $version ended");
        Log('cluster', "DXSpider V$version stopped");
+       dbgclose();
+       Logclose();
        unlink $lockfn;
+#      $SIG{__WARN__} = $SIG{__DIE__} =  sub {my $a = shift; cluck($a); };
        exit(0);
 }
 
@@ -197,21 +232,36 @@ sub process_inqueue
        
        my $data = $self->{data};
        my $dxchan = $self->{dxchan};
-       my ($sort, $call, $line) = $data =~ /^(\w)(\S+)\|(.*)$/;
+       my ($sort, $call, $line) = $data =~ /^(\w)([A-Z0-9\-]+)\|(.*)$/;
+       my $error;
+       
+       # the above regexp must work
+       return unless ($sort && $call && $line);
+       
+       # translate any crappy characters into hex characters 
+       if ($line =~ /[\x00-\x06\x08\x0a-\x1f\x7f-\xff]/o) {
+               $line =~ s/([\x00-\x1f\x7f-\xff])/uc sprintf("%%%02x",ord($1))/eg;
+               ++$error;
+#              dbg('chan', "<- $sort $call **CRAP**: $line");
+#              return;
+       }
        
        # do the really sexy console interface bit! (Who is going to do the TK interface then?)
        dbg('chan', "<- $sort $call $line\n") unless $sort eq 'D';
-       
+
        # handle A records
        my $user = $dxchan->user;
        if ($sort eq 'A' || $sort eq 'O') {
                $dxchan->start($line, $sort);  
        } elsif ($sort eq 'I') {
                die "\$user not defined for $call" if !defined $user;
-               
-               # normal input
-               $dxchan->normal($line);
-               
+
+               if ($error) {
+                       dbg('chan', "DROPPED with $error duff characters");
+               } else {
+                       # normal input
+                       $dxchan->normal($line);
+               }
                disconnect($dxchan) if ($dxchan->{state} eq 'bye');
        } elsif ($sort eq 'Z') {
                disconnect($dxchan);
@@ -241,6 +291,7 @@ sub uptime
 $starttime = $systime = time;
 
 # open the debug file, set various FHs to be unbuffered
+dbginit();
 foreach (@debug) {
        dbgadd($_);
 }
@@ -314,9 +365,10 @@ dbg('local', "Local::init error $@") if $@;
 
 # this, such as it is, is the main loop!
 print "orft we jolly well go ...\n";
+dbg('chan', "DXSpider version $version started...");
 for (;;) {
        my $timenow;
-       Msg->event_loop(1, 0.001);
+       Msg->event_loop(1, 0.1);
        $timenow = time;
        process_inqueue();                      # read in lines from the input queue and despatch them
        
@@ -329,6 +381,7 @@ for (;;) {
                DXCommandmode::process(); # process ongoing command mode stuff
                DXProt::process();              # process ongoing ak1a pcxx stuff
                DXConnect::process();
+               DXMsg::process();
                eval { 
                        Local::process();       # do any localised processing
                };