X-Git-Url: http://dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2Fcluster.pl;h=b67b9ff88e98b4c150a29d77dd61897ded2d0370;hb=465426f9282b990ec3462c63e5ca7613f92fb290;hp=8502cbaba6b3d44bc4bd8935a566fd6c042d8ce4;hpb=678715c76fa49f08acb30df0760b34407f49675c;p=spider.git diff --git a/perl/cluster.pl b/perl/cluster.pl index 8502cbab..b67b9ff8 100755 --- a/perl/cluster.pl +++ b/perl/cluster.pl @@ -14,12 +14,28 @@ 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'}; unshift @INC, "$root/perl"; # this IS the right way round! unshift @INC, "$root/local"; + + # try to create and lock a lockfile (this isn't atomic but + # should do for now + $lockfn = "$root/perl/cluster.lock"; # lock file name + if (-e $lockfn) { + open(CLLOCK, "$lockfn") or die "Can't open Lockfile ($lockfn) $!"; + my $pid = ; + chomp $pid; + die "Lockfile ($lockfn) and process $pid exist, another cluster running?" if kill 0, $pid; + close CLLOCK; + } + open(CLLOCK, ">$lockfn") or die "Can't open Lockfile ($lockfn) $!"; + print CLLOCK "$$\n"; + close CLLOCK; } use Msg; @@ -41,17 +57,24 @@ use Prefix; use Bands; use Geomag; use CmdAlias; +use Filter; use Local; +use DXDb; +use Data::Dumper; + +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.23"; # the version no of the software +$version = "1.35"; # 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 { @@ -70,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 @@ -86,9 +111,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)) { ; @@ -119,6 +151,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; } @@ -140,16 +173,55 @@ 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 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(); + + # 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); } @@ -158,6 +230,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 @@ -169,21 +242,28 @@ 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; + } # 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') { disconnect($dxchan); @@ -213,6 +293,7 @@ sub uptime $starttime = $systime = time; # open the debug file, set various FHs to be unbuffered +dbginit(); foreach (@debug) { dbgadd($_); } @@ -233,7 +314,7 @@ Bands::load(); # initialise User file system print "loading user file system ...\n"; -DXUser->init($userfn); +DXUser->init($userfn, 1); # start listening for incoming messages/connects print "starting listener ...\n"; @@ -274,6 +355,10 @@ DXMsg::clean_old(); print "reading cron jobs ...\n"; DXCron->init(); +# read in database descriptors +print "reading database descriptors ...\n"; +DXDb::load(); + # starting local stuff print "doing local initialisation ...\n"; eval { @@ -281,16 +366,15 @@ eval { }; dbg('local', "Local::init error $@") if $@; - - # print various flags #print "useful info - \$^D: $^D \$^W: $^W \$^S: $^S \$^P: $^P\n"; # 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 @@ -303,6 +387,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 }; @@ -312,5 +398,7 @@ for (;;) { last if --$decease <= 0; } } +cease(0); +exit(0);