X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2Fcluster.pl;h=be8380c1df0525c18ab69eb6b6d12b274c235344;hb=88665a2bed3b9ec9e97237938a95a045b2a21bb4;hp=fd78191479e9e171c0a6ab5003bf6ac664c89808;hpb=abab69b2a9860c96d8ef5a958056f239d8a18e98;p=spider.git diff --git a/perl/cluster.pl b/perl/cluster.pl index fd781914..be8380c1 100755 --- a/perl/cluster.pl +++ b/perl/cluster.pl @@ -14,6 +14,8 @@ require 5.004; # make sure that modules are searched in the order local then perl BEGIN { + umask 002; + # root of directory tree for this system $root = "/spider"; $root = $ENV{'DXSPIDER_ROOT'} if $ENV{'DXSPIDER_ROOT'}; @@ -55,18 +57,23 @@ use Prefix; use Bands; use Geomag; use CmdAlias; -use Local; +use Filter; +use DXDb; +use AnnTalk; + +use Data::Dumper; use Fcntl ':flock'; -use Carp; +use Local; package main; @inqueue = (); # the main input queue, an array of hashes $systime = 0; # the time now (in seconds) -$version = "1.24"; # the version no of the software +$version = "1.41"; # the version no of the software $starttime = 0; # the starting time of the cluster $lockfn = "cluster.lock"; # lock file name +@outstanding_connects = (); # list of outstanding connects # handle disconnections sub disconnect @@ -86,6 +93,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 @@ -94,22 +103,33 @@ sub rec my ($conn, $msg, $err) = @_; my $dxchan = DXChannel->get_by_cnum($conn); # get the dxconnnect object for this message - if (defined $err && $err) { - disconnect($dxchan) if defined $dxchan; + if (!defined $msg || (defined $err && $err)) { + if ($dxchan) { + $dxchan->disconnect; + } elsif ($conn) { + $conn->disconnect; + } return; } # 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 elsewhere in the cluster (and not a cluster) + + # is there one already connected to me - locally? my $user = DXUser->get($call); + if (DXChannel->get($call)) { + my $mess = DXM::msg($lang, ($user && $user->sort eq 'A') ? 'concluster' : 'conother', $call); + already_conn($conn, $call, $mess); + return; + } + + # is there one already connected elsewhere in the cluster? if ($user) { if (($user->sort eq 'A' || $call eq $myalias) && !DXCluster->get_exact($call)) { ; } else { - if (DXCluster->get($call) || DXChannel->get($call)) { + if (DXCluster->get_exact($call)) { my $mess = DXM::msg($lang, $user->sort eq 'A' ? 'concluster' : 'conother', $call); already_conn($conn, $call, $mess); return; @@ -117,7 +137,7 @@ sub rec } $user->{lang} = $main::lang if !$user->{lang}; # to autoupdate old systems } else { - if (DXCluster->get($call)) { + if (DXCluster->get_exact($call)) { my $mess = DXM::msg($lang, 'conother', $call); already_conn($conn, $call, $mess); return; @@ -135,6 +155,7 @@ sub rec # create the channel $dxchan = DXCommandmode->new($call, $conn, $user) if ($user->sort eq 'U'); $dxchan = DXProt->new($call, $conn, $user) if ($user->sort eq 'A'); + $dxchan = BBS->new($call, $conn, $user) if ($user->sort eq 'B'); die "Invalid sort of user on $call = $sort" if !$dxchan; } @@ -156,17 +177,49 @@ sub login sub cease { my $dxchan; + + $SIG{'TERM'} = 'IGNORE'; + $SIG{'INT'} = 'IGNORE'; eval { Local::finish(); # end local processing }; dbg('local', "Local::finish error $@") if $@; - + + # 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); + + # 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); + DXUser::finish(); + + # close all databases + DXDb::closeall; + + 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); } @@ -175,6 +228,7 @@ sub reap { $SIG{'CHLD'} = \&reap; my $cpid = wait; + @outstanding_connects = grep {$_->{pid} != $cpid} @outstanding_connects; } # this is where the input queue is dealt with and things are dispatched off to other parts of @@ -186,23 +240,31 @@ sub process_inqueue my $data = $self->{data}; my $dxchan = $self->{dxchan}; - my ($sort, $call, $line) = $data =~ /^(\w)(\S+)\|(.*)$/; + my ($sort, $call, $line) = $data =~ /^(\w)([^\|]+)\|(.*)$/; + 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; + } # 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); - disconnect($dxchan) if ($dxchan->{state} eq 'bye'); } elsif ($sort eq 'Z') { + $dxchan->conn(undef); disconnect($dxchan); } elsif ($sort eq 'D') { ; # ignored (an echo) @@ -230,6 +292,7 @@ sub uptime $starttime = $systime = time; # open the debug file, set various FHs to be unbuffered +dbginit(); foreach (@debug) { dbgadd($_); } @@ -238,22 +301,22 @@ STDOUT->autoflush(1); Log('cluster', "DXSpider V$version started"); # banner -print "DXSpider DX Cluster Version $version\nCopyright (c) 1998 Dirk Koopman G1TLH\n"; +dbg('err', "DXSpider DX Cluster Version $version", "Copyright (c) 1998-2000 Dirk Koopman G1TLH"); # load Prefixes -print "loading prefixes ...\n"; +dbg('err', "loading prefixes ..."); Prefix::load(); # load band data -print "loading band data ...\n"; +dbg('err', "loading band data ..."); Bands::load(); # initialise User file system -print "loading user file system ...\n"; -DXUser->init($userfn); +dbg('err', "loading user file system ..."); +DXUser->init($userfn, 1); # start listening for incoming messages/connects -print "starting listener ...\n"; +dbg('err', "starting listener ..."); Msg->new_server("$clusteraddr", $clusterport, \&login); # prime some signals @@ -275,7 +338,7 @@ Geomag->init(); Spot->init(); # initialise the protocol engine -print "reading in duplicate spot and WWV info ...\n"; +dbg('err', "reading in duplicate spot and WWV info ..."); DXProt->init(); @@ -283,31 +346,41 @@ DXProt->init(); DXNode->new(0, $mycall, 0, 1, $DXProt::myprot_version); # read in any existing message headers and clean out old crap -print "reading existing message headers ...\n"; +dbg('err', "reading existing message headers ..."); DXMsg->init(); DXMsg::clean_old(); # read in any cron jobs -print "reading cron jobs ...\n"; +dbg('err', "reading cron jobs ..."); DXCron->init(); +# read in database descriptors +dbg('err', "reading database descriptors ..."); +DXDb::load(); + # starting local stuff -print "doing local initialisation ...\n"; +dbg('err', "doing local initialisation ..."); eval { Local::init(); }; dbg('local', "Local::init error $@") if $@; # print various flags -#print "useful info - \$^D: $^D \$^W: $^W \$^S: $^S \$^P: $^P\n"; +#dbg('err', "seful info - \$^D: $^D \$^W: $^W \$^S: $^S \$^P: $^P"); # this, such as it is, is the main loop! -print "orft we jolly well go ...\n"; +dbg('err', "orft we jolly well go ..."); + +#open(DB::OUT, "|tee /tmp/aa"); + for (;;) { my $timenow; - Msg->event_loop(1, 0.001); +# $DB::trace = 1; + + Msg->event_loop(1, 0.1); $timenow = time; process_inqueue(); # read in lines from the input queue and despatch them +# $DB::trace = 0; # do timed stuff, ongoing processing happens one a second if ($timenow != $systime) { @@ -318,6 +391,8 @@ for (;;) { DXCommandmode::process(); # process ongoing command mode stuff DXProt::process(); # process ongoing ak1a pcxx stuff DXConnect::process(); + DXMsg::process(); + DXDb::process(); eval { Local::process(); # do any localised processing }; @@ -327,5 +402,7 @@ for (;;) { last if --$decease <= 0; } } +cease(0); +exit(0);