X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2Fcluster.pl;h=be8380c1df0525c18ab69eb6b6d12b274c235344;hb=88665a2bed3b9ec9e97237938a95a045b2a21bb4;hp=fdfff04f110abb24807561310a44543b76e96f24;hpb=4caf0d3cacae4ecb4995e7cedd725b953516792e;p=spider.git diff --git a/perl/cluster.pl b/perl/cluster.pl index fdfff04f..be8380c1 100755 --- a/perl/cluster.pl +++ b/perl/cluster.pl @@ -58,18 +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.31"; # 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 @@ -90,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 @@ -99,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; @@ -129,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; @@ -147,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; } @@ -177,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); @@ -189,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); @@ -200,13 +209,11 @@ 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 + DXDb::closeall; + dbg('chan', "DXSpider version $version ended"); Log('cluster', "DXSpider V$version stopped"); dbgclose(); @@ -221,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 @@ -232,16 +240,15 @@ 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 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; -# dbg('chan', "<- $sort $call **CRAP**: $line"); -# return; } # do the really sexy console interface bit! (Who is going to do the TK interface then?) @@ -253,12 +260,11 @@ sub process_inqueue $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) @@ -295,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 @@ -332,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(); @@ -340,32 +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('chan', "DXSpider version $version started..."); +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) { @@ -377,6 +392,7 @@ for (;;) { DXProt::process(); # process ongoing ak1a pcxx stuff DXConnect::process(); DXMsg::process(); + DXDb::process(); eval { Local::process(); # do any localised processing };