X-Git-Url: http://dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2Fcluster.pl;h=07fd6ab594630c0606a48bae4932f3abf38e4d2d;hb=9c8f43f26a7db08c4ff6ef2213c95c9f509abe36;hp=e64f558691c76ad1b18d8ed62863dca478dfbdd2;hpb=4b49fe47b37249d748f0a52fe2a778d4bc27232f;p=spider.git diff --git a/perl/cluster.pl b/perl/cluster.pl index e64f5586..07fd6ab5 100755 --- a/perl/cluster.pl +++ b/perl/cluster.pl @@ -10,17 +10,46 @@ # # +package main; + require 5.10.1; + use warnings; +use vars qw($root $is_win $systime $lockfn @inqueue $starttime $lockfn @outstanding_connects + $zombies @listeners $lang $myalias @debug $userfn $clusteraddr + $clusterport $mycall $decease $routeroot $me $reqreg $bumpexisting + $allowdxby $dbh $dsn $dbuser $dbpass $do_xml $systime_days $systime_daystart + $can_encode $maxconnect_user $maxconnect_node $idle_interval $log_flush_interval + $broadcast_debug + ); + +$lang = 'en'; # default language +$clusteraddr = '127.0.0.1'; # cluster tcp host address - used for things like console.pl +$clusterport = 27754; # cluster tcp port +$yes = 'Yes'; # visual representation of yes +$no = 'No'; # ditto for no +$user_interval = 11*60; # the interval between unsolicited prompts if no traffic + # make sure that modules are searched in the order local then perl BEGIN { umask 002; + $SIG{'__WARN__'} = sub { warn $_[0] if $DOWARN }; + + # take into account any local::lib that might be present + eval { + require local::lib; + }; + unless ($@) { +# import local::lib; + import local::lib qw(/spider/perl5lib); + } # root of directory tree for this system $root = "/spider"; $root = $ENV{'DXSPIDER_ROOT'} if $ENV{'DXSPIDER_ROOT'}; + unshift @INC, "$root/perl5lib" unless grep {$_ eq "$root/perl5lib"} @INC; unshift @INC, "$root/perl"; # this IS the right way round! unshift @INC, "$root/local"; @@ -29,18 +58,25 @@ BEGIN { die "$root/local doesn't exist, please RTFM" unless -d "$root/local"; die "$root/local/DXVars.pm doesn't exist, please RTFM" unless -e "$root/local/DXVars.pm"; - mkdir "$root/local_cmd", 0777 unless -d "$root/local_cmd"; + # create some directories + mkdir "$root/local_cmd", 02774 unless -d "$root/local_cmd"; + # locally stored data lives here + my $local_data = "$root/local_data"; + mkdir $local_data, 02774 unless -d $local_data; # try to create and lock a lockfile (this isn't atomic but # should do for now - $lockfn = "$root/local/cluster.lck"; # lock file name + $lockfn = "$root/local_data/cluster.lck"; # lock file name if (-w $lockfn) { open(CLLOCK, "$lockfn") or die "Can't open Lockfile ($lockfn) $!"; my $pid = ; if ($pid) { chomp $pid; - die "Lockfile ($lockfn) and process $pid exist, another cluster running?" if kill 0, $pid; + if (kill 0, $pid) { + warn "Lockfile ($lockfn) and process $pid exist, another cluster running?\n"; + exit 1; + } } unlink $lockfn; close CLLOCK; @@ -51,11 +87,18 @@ BEGIN { $is_win = ($^O =~ /^MS/ || $^O =~ /^OS-2/) ? 1 : 0; # is it Windows? $systime = time; + } +use DXVars; +use SysVar; + +# order here is important - DXDebug snarfs Carp et al so that Mojo errors go into the debug log +use Mojolicious 7.26; use Mojo::IOLoop; +$DOWARN = 1; -use DXVars; +use DXDebug; use Msg; use IntMsg; use Internet; @@ -63,7 +106,6 @@ use Listeners; use ExtMsg; use AGWConnect; use AGWMsg; -use DXDebug; use DXLog; use DXLogPrint; use DXUtil; @@ -108,26 +150,21 @@ use DXXml; use DXSql; use IsoTime; use BPQMsg; +use RBN; + use Data::Dumper; use IO::File; use Fcntl ':flock'; use POSIX ":sys_wait_h"; -use Version; use Web; -use Local; - -package main; +use vars qw($version $build $gitversion $gitbranch); use strict; -use vars qw(@inqueue $systime $starttime $lockfn @outstanding_connects - $zombies $root @listeners $lang $myalias @debug $userfn $clusteraddr - $clusterport $mycall $decease $is_win $routeroot $me $reqreg $bumpexisting - $allowdxby $dbh $dsn $dbuser $dbpass $do_xml $systime_days $systime_daystart - $can_encode $maxconnect_user $maxconnect_node $idle_interval $log_flush_interval - $broadcast_debug - ); + +use Local; + @inqueue = (); # the main input queue, an array of hashes $systime = 0; # the time now (in seconds) @@ -136,6 +173,16 @@ $starttime = 0; # the starting time of the cluster @listeners = (); # list of listeners $reqreg = 0; # 1 = registration required, 2 = deregister people $bumpexisting = 1; # 1 = allow new connection to disconnect old, 0 - don't allow it +our $allowmultiple = 0; # This is used in conjunction with $bumpexisting, in a rather weird way. +our $min_reconnection_rate = 5*60; # minimum value of seconds between connections per user to allow co-existing users +our $max_ssid = 15; # highest ssid to be searched for a spare one on multiple connections + +# If $allowmultiple > 0 and the $reconnection_rate is some value of seconds +# based on the average connection time calculated from the $user->conntimel entries / frequency is +# less than $reconnection_rate then we assume that there is more than one device (probably HRD) trying +# to connect "at once". In which case we probe for a spare SSID for a user callsign to allow up to +# $allowmultiple connections per callsign. + $allowdxby = 0; # 1 = allow "dx by ", 0 - don't allow it $maxconnect_user = 3; # the maximum no of concurrent connections a user can have at a time $maxconnect_node = 0; # Ditto but for nodes. In either case if a new incoming connection @@ -146,7 +193,9 @@ $log_flush_interval = 2; # interval to wait between log flushes our $ending; # signal that we are ending; our $broadcast_debug; # allow broadcasting of debug info down "enhanced" user connections - +our $clssecs; # the amount of cpu time the DXSpider process have consumed +our $cldsecs; # the amount of cpu time any child processes have consumed +our $allowslashcall; # Allow / in connecting callsigns (ie PA0/G1TLH, or even PA0/G1TLH/2) # send a message to call on conn and disconnect @@ -168,85 +217,162 @@ sub new_channel my ($sort, $call, $line) = DXChannel::decode_input(0, $msg); return unless defined $sort; - unless (is_callsign($call)) { - already_conn($conn, $call, DXM::msg($lang, "illcall", $call)); - return; - } - - # set up the basic channel info - # is there one already connected to me - locally? - my $user = DXUser::get_current($call); - my $dxchan = DXChannel::get($call); - if ($dxchan) { - if ($user && $user->is_node) { - already_conn($conn, $call, DXM::msg($lang, 'conother', $call, $main::mycall)); + my ($dxchan, $user); + + if (is_webcall($call) && $conn->isa('IntMsg')) { + my $newcall = find_next_webcall(); + unless ($newcall) { + already_conn($conn, $call, "Maximum no of web connected connects ($Web::maxssid) exceeded"); return; } - if ($bumpexisting) { - my $ip = $conn->peerhost || 'unknown'; - $dxchan->send_now('D', DXM::msg($lang, 'conbump', $call, $ip)); - LogDbg('DXCommand', "$call bumped off by $ip, disconnected"); - $dxchan->disconnect; - } else { - already_conn($conn, $call, DXM::msg($lang, 'conother', $call, $main::mycall)); + $call = $newcall; + $user = DXUser::get_current($call); + unless ($user) { + $user = DXUser->new($call); + $user->sort('W'); + $user->wantbeep(0); + $user->name('web'); + $user->qth('on the web'); + $user->homenode($main::mycall); + $user->lat($main::mylatitude); + $user->long($main::mylongitude); + $user->qra($main::mylocator); + } + $user->startt($main::systime); + $conn->conns($call); + $dxchan = Web->new($call, $conn, $user); + $dxchan->enhanced(1); + $dxchan->ve7cc(1); + $msg =~ s/^A#WEB|/A$call|/; + $conn->send_now("C$call"); + } else { + # "Normal" connections + unless (is_callsign($call)) { + already_conn($conn, $call, DXM::msg($lang, "illcall", $call)); return; } - } - # (fairly) politely disconnect people that are connected to too many other places at once - my $r = Route::get($call); - if ($conn->{sort} && $conn->{sort} =~ /^I/ && $r && $user) { - my @n = $r->parents; - my $m = $r->isa('Route::Node') ? $maxconnect_node : $maxconnect_user; - my $c = $user->maxconnect; - my $v; - $v = defined $c ? $c : $m; - if ($v && @n >= $v) { - my $nodes = join ',', @n; - LogDbg('DXCommand', "$call has too many connections ($v) at $nodes - disconnected"); - already_conn($conn, $call, DXM::msg($lang, 'contomany', $call, $v, $nodes)); - return; + # is he locked out ? + $user = DXUser::get_current($call); + $conn->conns($call); + my $basecall = $call; + $basecall =~ s/-\d+$//; # remember this for later multiple user processing + my $lock; + if ($user) { + # make sure we act on any locked status that the actual incoming call has. + $lock = $user->lockout; + } elsif ($allowmultiple && $call ne $basecall) { + # if we are allowing multiple connections and there is a basecall minus incoming ssid, use the basecall's lock status + $user = DXUser::get_current($basecall); + $lock = $user->lockout if $user; } - } - # is he locked out ? - my $basecall = $call; - $basecall =~ s/-\d+$//; - my $baseuser = DXUser::get_current($basecall); - my $lock = $user->lockout if $user; - if ($baseuser && $baseuser->lockout || $lock) { - if (!$user || !defined $lock || $lock) { - my $host = $conn->peerhost || "unknown"; - LogDbg('DXCommand', "$call on $host is locked out, disconnected"); + # now deal with the lock + if ($lock) { + my $host = $conn->peerhost; + LogDbg('', "$call on $host is locked out, disconnected"); $conn->disconnect; return; } - } - if ($user) { - $user->{lang} = $main::lang if !$user->{lang}; # to autoupdate old systems - } else { - $user = DXUser->new($call); - } + # set up the basic channel info for "Normal" Users + # is there one already connected to me - locally? - # create the channel - if ($user->is_node) { - $dxchan = DXProt->new($call, $conn, $user); - } elsif ($user->is_user) { - $dxchan = DXCommandmode->new($call, $conn, $user); -# } elsif ($user->is_bbs) { # there is no support so -# $dxchan = BBS->new($call, $conn, $user); # don't allow it!!! - } else { - die "Invalid sort of user on $call = $sort"; - } + $dxchan = DXChannel::get($call); + my $newcall = $call; + if ($dxchan) { + if ($user && $user->is_node) { + already_conn($conn, $call, DXM::msg($lang, 'conother', $call, $main::mycall)); + return; + } + if ($allowmultiple && $user->is_user) { + # determine whether we are a candidate, have we flip-flopped frequently enough? + my @lastconns = @{$user->connlist} if $user->connlist; + my $allow = 0; + if (@lastconns >= $DXUser::maxconnlist) { + $allow = $lastconns[-1]->[0] - $lastconns[0]->[0] < $min_reconnection_rate; + } + # search for a spare ssid + L1: for (my $count = $call =~ /-\d+$/?0:1; $allow && $count < $allowmultiple; ) { # remember we have one call already + my $lastid = 1; + for (; $lastid < $max_ssid && $count < $allowmultiple; ++$lastid) { + my $chan = DXChannel::get("$basecall-$lastid"); + if ($chan) { + ++$count; + next; + } + # we have a free call-ssid, save it + $newcall = "$basecall-$lastid"; + last L1; + } + } + } + + # handle "normal" (non-multiple) connections in the existing way + if ($call eq $newcall) { + if ($bumpexisting) { + my $ip = $dxchan->hostname; + $dxchan->send_now('D', DXM::msg($lang, 'conbump', $call, $ip)); + LogDbg('', "$call bumped off by $ip, disconnected"); + $dxchan->disconnect; + } else { + already_conn($conn, $call, DXM::msg($lang, 'conother', $call, $main::mycall)); + return; + } + } + + # make sure that the conn has the (possibly) new callsign + $conn->conns($newcall); + $msg =~ s/$call/$newcall/; + } + + + # (fairly) politely disconnect people that are connected to too many other places at once + my $r = Route::get($call); + if ($conn->{sort} && $conn->{sort} =~ /^I/ && $r && $user) { + my @n = $r->parents; + my $m = $r->isa('Route::Node') ? $maxconnect_node : $maxconnect_user; + my $c = $user->maxconnect; + my $v; + $v = defined $c ? $c : $m; + if ($v && @n >= $v+$allowmultiple) { + my $nodes = join ',', @n; + LogDbg('', "$call has too many connections ($v) at $nodes - disconnected"); + already_conn($conn, $call, DXM::msg($lang, 'contomany', $call, $v, $nodes)); + return; + } + } + + if ($user) { + $user->{lang} = $main::lang if !$user->{lang}; # to autoupdate old systems + } else { + $user = DXUser->new($call); + } - # check that the conn has a callsign - $conn->conns($call) if $conn->isa('IntMsg'); + # create the channel + # NOTE we are SHARING the same $user if $allowmultiple is > 1 + + $user->startt($systime); # mark the start time of this connection + if ($user->is_node) { + $dxchan = DXProt->new($call, $conn, $user); + } elsif ($user->is_rbn) { + $dxchan = RBN->new($newcall, $conn, $user); + } elsif ($user->is_user) { + $dxchan = DXCommandmode->new($newcall, $conn, $user); + } else { + die "Invalid sort of user on $call = $sort"; + } + } + # set callbacks - $conn->set_error(sub {my $err = shift; LogDbg('DXCommand', "Comms error '$err' received for call $dxchan->{call}"); $dxchan->disconnect(1);}); + $conn->set_error(sub {my $err = shift; LogDbg('', "Comms error '$err' received for call $dxchan->{call}"); $dxchan->disconnect(1);}); $conn->set_on_eof(sub {$dxchan->disconnect}); $conn->set_rproc(sub {my ($conn,$msg) = @_; $dxchan->rec($msg);}); + if ($sort eq 'W') { + $dxchan->enhanced(1); + $dxchan->sort('W'); + } $dxchan->rec($msg); } @@ -272,7 +398,6 @@ sub cease $SIG{'INT'} = 'IGNORE'; } - DXUser::sync; if (defined &Local::finish) { eval { @@ -290,6 +415,7 @@ sub cease UDPMsg::finish(); # end everything else + RBN::finish(); DXUser::finish(); DXDupe::finish(); @@ -301,7 +427,7 @@ sub cease $l->close_server; } - LogDbg('cluster', "DXSpider V$version, build $build (git: $gitversion) ended"); + LogDbg('cluster', "DXSpider V$version, build $build (git: $gitbranch/$gitversion) ended"); dbg("bye bye everyone - bye bye"); dbgclose(); Logclose(); @@ -342,77 +468,6 @@ sub AGWrestart AGWMsg::init(\&new_channel); } -our $io_disconnected; - -sub idle_loop -{ - my $timenow = time; - - BPQMsg::process(); -# DXChannel::process(); - - # $DB::trace = 0; - - # do timed stuff, ongoing processing happens one a second - if ($timenow != $systime) { - reap() if $zombies; - $systime = $timenow; - my $days = int ($systime / 86400); - if ($systime_days != $days) { - $systime_days = $days; - $systime_daystart = $days * 86400; - } - IsoTime::update($systime); - DXCron::process(); # do cron jobs - DXCommandmode::process(); # process ongoing command mode stuff - DXXml::process(); - DXProt::process(); # process ongoing ak1a pcxx stuff - DXConnect::process(); - DXMsg::process(); - DXDb::process(); - DXUser::process(); - DXDupe::process(); - DXCron::process(); # do cron jobs - IsoTime::update($systime); - DXProt::process(); # process ongoing ak1a pcxx stuff - DXConnect::process(); - DXUser::process(); - AGWMsg::process(); - - Timer::handler(); - DXLog::flushall(); - } - - if (defined &Local::process) { - eval { - Local::process(); # do any localised processing - }; - dbg("Local::process error $@") if $@; - } - - while ($ending) { - my $dxchan; - - dbg("DXSpider Ending $ending"); - - unless ($io_disconnected++) { - - # disconnect users - foreach $dxchan (DXChannel::get_all_users) { - $dxchan->disconnect; - } - - # disconnect nodes - foreach $dxchan (DXChannel::get_all_nodes) { - next if $dxchan == $main::me; - $dxchan->disconnect(2); - } - $main::me->disconnect; - } - - Mojo::IOLoop->stop if --$ending <= 0; - } -} sub setup_start { @@ -433,12 +488,15 @@ sub setup_start } # open the debug file, set various FHs to be unbuffered - dbginit($broadcast_debug ? \&DXCommandmode::broadcast_debug : undef); + dbginit(undef, $broadcast_debug ? \&DXCommandmode::broadcast_debug : undef); foreach (@debug) { dbgadd($_); } STDOUT->autoflush(1); + # log our path + dbg "Perl path: " . join(':', @INC); + # try to load the database if (DXSql::init($dsn)) { $dbh = DXSql->new($dsn); @@ -455,6 +513,8 @@ sub setup_start import Encode; $can_encode = 1; } + $gitbranch = 'none'; + $gitversion = 'none'; eval { require Git; }; unless ($@) { import Git; @@ -465,10 +525,21 @@ sub setup_start my $desc = $repo->command_oneline(['describe', '--long'], STDERR => 0); if ($desc) { my ($v, $s, $b, $g) = $desc =~ /^([\d.]+)(?:\.(\d+))?-(\d+)-g([0-9a-f]+)/; + $s ||= ''; + dbg("Git: $desc") if isdbg('git'); + dbg("Git: V=$v S=$s B=$b g=$g") if isdbg('git'); $version = $v; $build = $b || 0; $gitversion = "$g\[r]"; } + my @branch = $repo->command([qw{branch}], STDERR=>0); + for (@branch) { + my ($star, $b) = split /\s+/; + if ($star eq '*') { + $gitbranch = $b; + last; + } + } } } $SIG{__DIE__} = $w; @@ -480,7 +551,7 @@ sub setup_start # banner my ($year) = (gmtime)[5]; $year += 1900; - LogDbg('cluster', "DXSpider V$version, build $build (git: $gitversion) started"); + LogDbg('cluster', "DXSpider V$version, build $build (git: $gitbranch/$gitversion) started"); dbg("Copyright (c) 1998-$year Dirk Koopman G1TLH"); # load Prefixes @@ -495,15 +566,28 @@ sub setup_start # initialise User file system dbg("loading user file system ..."); - DXUser->init($userfn, 1); + DXUser::init(4); # version 4 == json format + + Filter::init(); # doesn't do much, but has to be done + # look for the sysop and the alias user and complain if they aren't there { die "\$myalias \& \$mycall are the same ($mycall)!, they must be different (hint: make \$mycall = '${mycall}-2';). Oh and don't forget to rerun create_sysop.pl!" if $mycall eq $myalias; my $ref = DXUser::get($mycall); die "$mycall missing, run the create_sysop.pl script and please RTFM" unless $ref && $ref->priv == 9; + my $oldsort = $ref->sort; + if ($oldsort ne 'S') { + $ref->sort('S'); + dbg("Resetting node type from $oldsort -> DXSpider ('S')"); + } $ref = DXUser::get($myalias); die "$myalias missing, run the create_sysop.pl script and please RTFM" unless $ref && $ref->priv == 9; + $oldsort = $ref->sort; + if ($oldsort ne 'U') { + $ref->sort('U'); + dbg("Resetting sysop user type from $oldsort -> User ('U')"); + } } # start listening for incoming messages/connects @@ -608,6 +692,9 @@ sub setup_start dbg("reading database descriptors ..."); DXDb::load(); + dbg("starting RBN ..."); + RBN::init(); + # starting local stuff dbg("doing local initialisation ..."); QSL::init(1); @@ -627,14 +714,125 @@ sub setup_start #open(DB::OUT, "|tee /tmp/aa"); } +our $io_disconnected; + +sub idle_loop +{ + BPQMsg::process(); +# DXCommandmode::process(); # process ongoing command mode stuff +# DXProt::process(); # process ongoing ak1a pcxx stuff + + if (defined &Local::process) { + eval { + Local::process(); # do any localised processing + }; + dbg("Local::process error $@") if $@; + } + + while ($ending) { + my $dxchan; + + dbg("DXSpider Ending $ending"); + + unless ($io_disconnected++) { + + # disconnect users + foreach $dxchan (DXChannel::get_all_users) { + $dxchan->disconnect; + } + + # disconnect nodes + foreach $dxchan (DXChannel::get_all_nodes) { + next if $dxchan == $main::me; + $dxchan->disconnect(2); + } + $main::me->disconnect; + } + + Mojo::IOLoop->stop if --$ending <= 0; + } +} + +sub per_sec +{ + my $timenow = time; + + reap() if $zombies; + $systime = $timenow; + my $days = int ($systime / 86400); + if ($systime_days != $days) { + $systime_days = $days; + $systime_daystart = $days * 86400; + } + IsoTime::update($systime); + DXCommandmode::process(); # process ongoing command mode stuff + DXProt::process(); # process ongoing ak1a pcxx stuff + DXXml::process(); + DXConnect::process(); + DXMsg::process(); + DXDb::process(); + DXUser::process(); + DXDupe::process(); + IsoTime::update($systime); + DXConnect::process(); + DXUser::process(); + AGWMsg::process(); + DXCron::process(); # do cron jobs + RBN::process(); + + Timer::handler(); + DXLog::flushall(); +} + +sub per_10_sec +{ + +} + +sub per_minute +{ + RBN::per_minute(); +} + +sub per_10_minute +{ + RBN::per_10_minute(); +} + +sub per_hour +{ + RBN::per_hour(); +} + +sub per_day +{ + +} + +sub start_node +{ + dbg("Before Web::start_node"); + + Mojo::IOLoop->start unless Mojo::IOLoop->is_running; + + dbg("After Web::start_node"); +} setup_start(); my $main_loop = Mojo::IOLoop->recurring($idle_interval => \&idle_loop); my $log_flush_loop = Mojo::IOLoop->recurring($log_flush_interval => \&DXLog::flushall); +my $cpusecs_loop = Mojo::IOLoop->recurring(5 => sub {my @t = times; $clssecs = $t[0]+$t[1]; $cldsecs = $t[2]+$t[3]}); +my $persec = Mojo::IOLoop->recurring(1 => \&per_sec); +my $per10sec = Mojo::IOLoop->recurring(10 => \&per_10_sec); +my $permin = Mojo::IOLoop->recurring(60 => \&per_minute); +my $per10min = Mojo::IOLoop->recurring(600 => \&per_10_minute); +my $perhour = Mojo::IOLoop->recurring(3600 => \&per_hour); +my $perday = Mojo::IOLoop->recurring(86400 => \&per_day); -Web::start_node(); +start_node(); cease(0); + exit(0);