X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2Fcluster.pl;h=be8380c1df0525c18ab69eb6b6d12b274c235344;hb=88665a2bed3b9ec9e97237938a95a045b2a21bb4;hp=36c2f8bd027005b8ae6a16db4cfba6fe5d1e2477;hpb=0e84728de35d2dadbb9d624614a471b68ae9eef6;p=spider.git diff --git a/perl/cluster.pl b/perl/cluster.pl index 36c2f8bd..be8380c1 100755 --- a/perl/cluster.pl +++ b/perl/cluster.pl @@ -58,19 +58,22 @@ use Bands; use Geomag; use CmdAlias; use Filter; -use Local; use DXDb; +use AnnTalk; + +use Data::Dumper; use Fcntl ':flock'; -use Carp qw(cluck); +use Local; package main; @inqueue = (); # the main input queue, an array of hashes $systime = 0; # the time now (in seconds) -$version = "1.35"; # 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 @@ -91,7 +94,7 @@ sub already_conn dbg('chan', "-> Z $call bye\n"); $conn->send_now("Z$call|bye"); # this will cause 'client' to disconnect sleep(1); - $conn->disconnect; + $conn->disconnect(); } # handle incoming messages @@ -100,29 +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 to me ? + + # is there one already connected to me - locally? my $user = DXUser->get($call); if (DXChannel->get($call)) { - my $mess = DXM::msg($lang, $user->sort eq 'A' ? 'concluster' : 'conother', $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 (and not a cluster) + # 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; @@ -130,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; @@ -148,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; } @@ -178,9 +186,9 @@ sub cease }; dbg('local', "Local::finish error $@") if $@; - # disconnect users + # disconnect nodes foreach $dxchan (DXChannel->get_all()) { - next if $dxchan->is_ak1a; + next unless $dxchan->is_ak1a; disconnect($dxchan) unless $dxchan == $DXProt::me; } Msg->event_loop(1, 0.05); @@ -190,9 +198,9 @@ sub cease Msg->event_loop(1, 0.05); Msg->event_loop(1, 0.05); - # disconnect nodes + # disconnect users foreach $dxchan (DXChannel->get_all()) { - next unless $dxchan->is_ak1a; + next if $dxchan->is_ak1a; disconnect($dxchan) unless $dxchan == $DXProt::me; } Msg->event_loop(1, 0.05); @@ -201,12 +209,6 @@ sub cease 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(); # close all databases @@ -226,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 @@ -237,7 +240,7 @@ sub process_inqueue my $data = $self->{data}; my $dxchan = $self->{dxchan}; - my ($sort, $call, $line) = $data =~ /^(\w)([A-Z0-9\-]+)\|(.*)$/; + my ($sort, $call, $line) = $data =~ /^(\w)([^\|]+)\|(.*)$/; my $error; # the above regexp must work @@ -261,6 +264,7 @@ sub process_inqueue $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) @@ -297,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"; +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 @@ -334,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(); @@ -342,36 +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 -print "reading database descriptors ...\n"; +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('chan', "DXSpider version $version started..."); +dbg('err', "orft we jolly well go ..."); + +#open(DB::OUT, "|tee /tmp/aa"); + for (;;) { my $timenow; +# $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) {