]> dxcluster.org Git - spider.git/blob - perl/cluster.pl
*** empty log message ***
[spider.git] / perl / cluster.pl
1 #!/usr/bin/perl -w
2 #
3 # This is the DX cluster 'daemon'. It sits in the middle of its little
4 # web of client routines sucking and blowing data where it may.
5 #
6 # Hence the name of 'spider' (although it may become 'dxspider')
7 #
8 # Copyright (c) 1998 Dirk Koopman G1TLH
9 #
10 # $Id$
11
12
13 require 5.004;
14
15 # make sure that modules are searched in the order local then perl
16 BEGIN {
17         umask 002;
18         
19         # root of directory tree for this system
20         $root = "/spider"; 
21         $root = $ENV{'DXSPIDER_ROOT'} if $ENV{'DXSPIDER_ROOT'};
22         
23         unshift @INC, "$root/perl";     # this IS the right way round!
24         unshift @INC, "$root/local";
25
26         # do some validation of the input
27         die "The directory $root doesn't exist, please RTFM" unless -d $root;
28         die "$root/local doesn't exist, please RTFM" unless -d "$root/local";
29         die "$root/local/DXVars.pm doesn't exist, please RTFM" unless -e "$root/local/DXVars.pm";
30         
31         mkdir "$root/local_cmd", 0777 unless -d "$root/local_cmd";
32         
33
34         # try to create and lock a lockfile (this isn't atomic but 
35         # should do for now
36         $lockfn = "$root/perl/cluster.lck";       # lock file name
37         if (-e $lockfn) {
38                 open(CLLOCK, "$lockfn") or die "Can't open Lockfile ($lockfn) $!";
39                 my $pid = <CLLOCK>;
40                 chomp $pid;
41                 die "Lockfile ($lockfn) and process $pid exist, another cluster running?" if kill 0, $pid;
42                 close CLLOCK;
43         }
44         open(CLLOCK, ">$lockfn") or die "Can't open Lockfile ($lockfn) $!";
45         print CLLOCK "$$\n";
46         close CLLOCK;
47
48         $is_win = ($^O =~ /^MS/ || $^O =~ /^OS-2/) ? 1 : 0; # is it Windows?
49         $systime = time;
50 }
51
52 use DXVars;
53 use Msg;
54 use IntMsg;
55 use Internet;
56 use Listeners;
57 use ExtMsg;
58 use AGWConnect;
59 use AGWMsg;
60 use DXDebug;
61 use DXLog;
62 use DXLogPrint;
63 use DXUtil;
64 use DXChannel;
65 use DXUser;
66 use DXM;
67 use DXCommandmode;
68 use DXProtVars;
69 use DXProtout;
70 use DXProt;
71 use QXProt;
72 use DXMsg;
73 use DXCron;
74 use DXConnect;
75 use DXBearing;
76 use DXDb;
77 use DXHash;
78 use DXDupe;
79 use Script;
80 use Prefix;
81 use Spot;
82 use Bands;
83 use Keps;
84 use Minimuf;
85 use Sun;
86 use Geomag;
87 use CmdAlias;
88 use Filter;
89 use AnnTalk;
90 use BBS;
91 use WCY;
92 use BadWords;
93 use Timer;
94 use Route;
95 use Route::Node;
96 use Route::User;
97 use Editable;
98 use Mrtg;
99 use USDB;
100 use UDPMsg;
101
102 use Data::Dumper;
103 use IO::File;
104 use Fcntl ':flock'; 
105 use POSIX ":sys_wait_h";
106
107 use Local;
108
109 package main;
110
111 use strict;
112 use vars qw(@inqueue $systime $version $starttime $lockfn @outstanding_connects 
113                         $zombies $root @listeners $lang $myalias @debug $userfn $clusteraddr 
114                         $clusterport $mycall $decease $is_win $routeroot $me $reqreg
115                    );
116
117 @inqueue = ();                                  # the main input queue, an array of hashes
118 $systime = 0;                                   # the time now (in seconds)
119 $version = "1.51";                              # the version no of the software
120 $starttime = 0;                 # the starting time of the cluster   
121 #@outstanding_connects = ();     # list of outstanding connects
122 @listeners = ();                                # list of listeners
123 $reqreg = 0;                                    # 1 = registration required, 2 = deregister people
124
125 use vars qw($VERSION $BRANCH $build $branch);
126 $VERSION = sprintf( "%d.%03d", q$Revision$ =~ /(\d+)\.(\d+)/ );
127 $BRANCH = sprintf( "%d.%03d", q$Revision$ =~ /\d+\.\d+\.(\d+)\.(\d+)/  || (0,0));
128 $main::build += 6;                              # add an offset to make it bigger than last system
129 $main::build += $VERSION;
130 $main::branch += $BRANCH;
131
132       
133 # send a message to call on conn and disconnect
134 sub already_conn
135 {
136         my ($conn, $call, $mess) = @_;
137
138         $conn->disable_read(1);
139         dbg("-> D $call $mess\n") if isdbg('chan'); 
140         $conn->send_now("D$call|$mess");
141         sleep(2);
142         $conn->disconnect;
143 }
144
145 sub error_handler
146 {
147         my $dxchan = shift;
148         $dxchan->{conn}->set_error(undef) if exists $dxchan->{conn};
149         $dxchan->disconnect(1);
150 }
151
152 # handle incoming messages
153 sub new_channel
154 {
155         my ($conn, $msg) = @_;
156         my ($sort, $call, $line) = DXChannel::decode_input(0, $msg);
157         return unless defined $sort;
158
159         unless (is_callsign($call)) {
160                 already_conn($conn, $call, DXM::msg($lang, "illcall", $call));
161                 return;
162         }
163
164         # set up the basic channel info
165         # is there one already connected to me - locally? 
166         my $user = DXUser->get_current($call);
167         my $dxchan = DXChannel->get($call);
168         if ($dxchan) {
169                 my $mess = DXM::msg($lang, ($user && $user->is_node) ? 'concluster' : 'conother', $call, $main::mycall);
170                 already_conn($conn, $call, $mess);
171                 return;
172         }
173
174         # is he locked out ?
175         my $basecall = $call;
176         $basecall =~ s/-\d+$//;
177         my $baseuser = DXUser->get_current($basecall);
178         my $lock = $user->lockout if $user;
179         if ($baseuser && $baseuser->lockout || $lock) {
180                 if (!$user || !defined $lock || $lock) {
181                         my $host = $conn->{peerhost} || "unknown";
182                         Log('DXCommand', "$call on $host is locked out, disconnected");
183                         $conn->disconnect;
184                         return;
185                 }
186         }
187         
188         if ($user) {
189                 $user->{lang} = $main::lang if !$user->{lang}; # to autoupdate old systems
190         } else {
191                 $user = DXUser->new($call);
192         }
193         
194
195         # create the channel
196         if ($user->wantnp) {
197                 if ($user->passphrase && $main::me->user->passphrase) {
198                         $dxchan = QXProt->new($call, $conn, $user);
199                 } else {
200                         unless ($user->passphrase) {
201                                 Log('DXCommand', "$call using NP but has no passphrase");
202                                 dbg("$call using NP but has no passphrase");
203                         }
204                         unless ($main::me->user->passphrase) {
205                                 Log('DXCommand', "$main::mycall using NP but has no passphrase");
206                                 dbg("$main::mycall using NP but has no passphrase");
207                         }
208                         already_conn($conn, $call, "Need to exchange passphrases");
209                         return;
210                 }
211         } elsif ($user->is_node) {
212                 $dxchan = DXProt->new($call, $conn, $user);
213         } elsif ($user->is_user) {
214                 $dxchan = DXCommandmode->new($call, $conn, $user);
215         } elsif ($user->is_bbs) {
216                 $dxchan = BBS->new($call, $conn, $user);
217         } else {
218                 die "Invalid sort of user on $call = $sort";
219         }
220
221         # check that the conn has a callsign
222         $conn->conns($call) if $conn->isa('IntMsg');
223
224         # set callbacks
225         $conn->set_error(sub {error_handler($dxchan)});
226         $conn->set_rproc(sub {my ($conn,$msg) = @_; rec($dxchan, $conn, $msg);});
227         rec($dxchan, $conn, $msg);
228 }
229
230 sub rec 
231 {
232         my ($dxchan, $conn, $msg) = @_;
233         
234         # queue the message and the channel object for later processing
235         if (defined $msg) {
236                 my $self = bless {}, "inqueue";
237                 $self->{dxchan} = $dxchan;
238                 $self->{data} = $msg;
239                 push @inqueue, $self;
240         }
241 }
242
243 # remove any outstanding entries on the inqueue after a disconnection (usually)
244 sub clean_inqueue
245 {
246         my $dxchan = shift;
247         @inqueue = grep {$_->{dxchan} != $dxchan} @inqueue;
248 }
249
250 sub login
251 {
252         return \&new_channel;
253 }
254
255 # cease running this program, close down all the connections nicely
256 sub cease
257 {
258         my $dxchan;
259
260         unless ($is_win) {
261                 $SIG{'TERM'} = 'IGNORE';
262                 $SIG{'INT'} = 'IGNORE';
263         }
264         
265         DXUser::sync;
266
267         eval {
268                 Local::finish();   # end local processing
269         };
270         dbg("Local::finish error $@") if $@;
271
272         # disconnect nodes
273         foreach $dxchan (DXChannel->get_all_nodes) {
274             $dxchan->disconnect(2) unless $dxchan == $main::me;
275         }
276         Msg->event_loop(100, 0.01);
277
278         # disconnect users
279         foreach $dxchan (DXChannel->get_all_users) {
280                 $dxchan->disconnect;
281         }
282
283         # disconnect AGW
284         AGWMsg::finish();
285
286         # disconnect UDP customers
287         UDPMsg::finish();
288
289         # end everything else
290         Msg->event_loop(100, 0.01);
291         DXUser::finish();
292         DXDupe::finish();
293
294         # close all databases
295         DXDb::closeall;
296
297         # close all listeners
298         foreach my $l (@listeners) {
299                 $l->close_server;
300         }
301
302         dbg("DXSpider version $version, build $build ended") if isdbg('chan');
303         Log('cluster', "DXSpider V$version, build $build ended");
304         dbgclose();
305         Logclose();
306         unlink $lockfn;
307 #       $SIG{__WARN__} = $SIG{__DIE__} =  sub {my $a = shift; cluck($a); };
308         exit(0);
309 }
310
311 # the reaper of children
312 sub reap
313 {
314         my $cpid;
315         while (($cpid = waitpid(-1, WNOHANG)) > 0) {
316                 dbg("cpid: $cpid") if isdbg('reap');
317 #               Msg->pid_gone($cpid);
318                 $zombies-- if $zombies > 0;
319         }
320         dbg("cpid: $cpid") if isdbg('reap');
321 }
322
323 # this is where the input queue is dealt with and things are dispatched off to other parts of
324 # the cluster
325 sub process_inqueue
326 {
327         while (@inqueue) {
328                 my $self = shift @inqueue;
329                 return if !$self;
330
331                 my $data = $self->{data};
332                 my $dxchan = $self->{dxchan};
333                 my $error;
334                 my ($sort, $call, $line) = DXChannel::decode_input($dxchan, $data);
335                 return unless defined $sort;
336         
337                 # do the really sexy console interface bit! (Who is going to do the TK interface then?)
338                 dbg("<- $sort $call $line\n") if $sort ne 'D' && isdbg('chan');
339                 if ($self->{disconnecting}) {
340                         dbg('In disconnection, ignored');
341                         next;
342                 }
343                 
344                 # handle A records
345                 my $user = $dxchan->user;
346                 if ($sort eq 'A' || $sort eq 'O') {
347                         $dxchan->start($line, $sort);  
348                 } elsif ($sort eq 'I') {
349                         die "\$user not defined for $call" if !defined $user;
350
351                         # normal input
352                         $dxchan->normal($line);
353                 } elsif ($sort eq 'Z') {
354                         $dxchan->disconnect;
355                 } elsif ($sort eq 'D') {
356                         ;                                       # ignored (an echo)
357                 } elsif ($sort eq 'G') {
358                         $dxchan->enhanced($line);
359                 } else {
360                         print STDERR atime, " Unknown command letter ($sort) received from $call\n";
361                 }
362         }
363 }
364
365 sub uptime
366 {
367         my $t = $systime - $starttime;
368         my $days = int $t / 86400;
369         $t -= $days * 86400;
370         my $hours = int $t / 3600;
371         $t -= $hours * 3600;
372         my $mins = int $t / 60;
373         return sprintf "%d %02d:%02d", $days, $hours, $mins;
374 }
375
376 sub AGWrestart
377 {
378         AGWMsg::init(\&new_channel);
379 }
380
381 #############################################################
382 #
383 # The start of the main line of code 
384 #
385 #############################################################
386
387 $starttime = $systime = time;
388 $lang = 'en' unless $lang;
389
390 # open the debug file, set various FHs to be unbuffered
391 dbginit(\&DXCommandmode::broadcast_debug);
392 foreach (@debug) {
393         dbgadd($_);
394 }
395 STDOUT->autoflush(1);
396
397 # calculate build number
398 $build += $main::version;
399 $build = "$build.$branch" if $branch;
400
401 Log('cluster', "DXSpider V$version, build $build started");
402
403 # banner
404 dbg("Copyright (c) 1998-2002 Dirk Koopman G1TLH");
405 dbg("DXSpider Version $version, build $build started");
406
407 # load Prefixes
408 dbg("loading prefixes ...");
409 my $r = Prefix::init();
410 confess $r if $r;
411 dbg(USDB::init());
412
413 # load band data
414 dbg("loading band data ...");
415 Bands::load();
416
417 # initialise User file system
418 dbg("loading user file system ..."); 
419 DXUser->init($userfn, 1);
420
421 # look for the sysop and the alias user and complain if they aren't there
422 {
423         my $ref = DXUser->get($mycall);
424         die "$mycall missing, run the create_sysop.pl script and please RTFM" unless $ref && $ref->priv == 9;
425         $ref = DXUser->get($myalias);
426         die "$myalias missing, run the create_sysop.pl script and please RTFM" unless $ref && $ref->priv == 9;
427 }
428
429 # start listening for incoming messages/connects
430 dbg("starting listeners ...");
431 my $conn = IntMsg->new_server($clusteraddr, $clusterport, \&login);
432 $conn->conns("Server $clusteraddr/$clusterport");
433 push @listeners, $conn;
434 dbg("Internal port: $clusteraddr $clusterport");
435 foreach my $l (@main::listen) {
436         $conn = ExtMsg->new_server($l->[0], $l->[1], \&login);
437         $conn->conns("Server $l->[0]/$l->[1]");
438         push @listeners, $conn;
439         dbg("External Port: $l->[0] $l->[1]");
440 }
441
442 dbg("AGW Listener") if $AGWMsg::enable;
443 AGWrestart();
444
445 dbg("UDP Listener") if $UDPMsg::enable;
446 UDPMsg::init(\&new_channel);
447
448 # load bad words
449 dbg("load badwords: " . (BadWords::load or "Ok"));
450
451 # prime some signals
452 unless ($DB::VERSION) {
453         $SIG{INT} = $SIG{TERM} = sub { $decease = 1 };
454 }
455
456 unless ($is_win) {
457         $SIG{HUP} = 'IGNORE';
458         $SIG{CHLD} = sub { $zombies++ };
459         
460         $SIG{PIPE} = sub {      dbg("Broken PIPE signal received"); };
461         $SIG{IO} = sub {        dbg("SIGIO received"); };
462         $SIG{WINCH} = $SIG{STOP} = $SIG{CONT} = 'IGNORE';
463         $SIG{KILL} = 'DEFAULT';     # as if it matters....
464
465         # catch the rest with a hopeful message
466         for (keys %SIG) {
467                 if (!$SIG{$_}) {
468                         #               dbg("Catching SIG $_") if isdbg('chan');
469                         $SIG{$_} = sub { my $sig = shift;       DXDebug::confess("Caught signal $sig");  }; 
470                 }
471         }
472 }
473
474 # start dupe system
475 dbg("Starting Dupe system");
476 DXDupe::init();
477
478 # read in system messages
479 dbg("Read in Messages");
480 DXM->init();
481
482 # read in command aliases
483 dbg("Read in Aliases");
484 CmdAlias->init();
485
486 # initialise the Geomagnetic data engine
487 dbg("Start WWV");
488 Geomag->init();
489 dbg("Start WCY");
490 WCY->init();
491
492 # initial the Spot stuff
493 dbg("Starting DX Spot system");
494 Spot->init();
495
496 # initialise the protocol engine
497 dbg("Start Protocol Engines ...");
498 DXProt->init();
499 QXProt->init();
500
501 # put in a DXCluster node for us here so we can add users and take them away
502 $routeroot = Route::Node->new($mycall, $version*100+5300, Route::here($main::me->here)|Route::conf($main::me->conf));
503
504 # make sure that there is a routing OUTPUT node default file
505 #unless (Filter::read_in('route', 'node_default', 0)) {
506 #       my $dxcc = $main::me->dxcc;
507 #       $Route::filterdef->cmd($main::me, 'route', 'accept', "node_default call $mycall" );
508 #}
509
510 # read in any existing message headers and clean out old crap
511 dbg("reading existing message headers ...");
512 DXMsg->init();
513 DXMsg::clean_old();
514
515 # read in any cron jobs
516 dbg("reading cron jobs ...");
517 DXCron->init();
518
519 # read in database descriptors
520 dbg("reading database descriptors ...");
521 DXDb::load();
522
523 # starting local stuff
524 dbg("doing local initialisation ...");
525 eval {
526         Local::init();
527 };
528 dbg("Local::init error $@") if $@;
529
530 # this, such as it is, is the main loop!
531 dbg("orft we jolly well go ...");
532 my $script = new Script "startup";
533 $script->run($main::me) if $script;
534
535 #open(DB::OUT, "|tee /tmp/aa");
536
537 for (;;) {
538 #       $DB::trace = 1;
539         
540         Msg->event_loop(10, 0.010);
541         my $timenow = time;
542         process_inqueue();                      # read in lines from the input queue and despatch them
543 #       $DB::trace = 0;
544         
545         # do timed stuff, ongoing processing happens one a second
546         if ($timenow != $systime) {
547                 reap if $zombies;
548                 $systime = $timenow;
549                 DXCron::process();      # do cron jobs
550                 DXCommandmode::process(); # process ongoing command mode stuff
551                 DXProt::process();              # process ongoing ak1a pcxx stuff
552                 QXProt::process();
553                 DXConnect::process();
554                 DXMsg::process();
555                 DXDb::process();
556                 DXUser::process();
557                 DXDupe::process();
558                 AGWMsg::process();
559                                 
560                 eval { 
561                         Local::process();       # do any localised processing
562                 };
563                 dbg("Local::process error $@") if $@;
564         }
565         if ($decease) {
566                 last if --$decease <= 0;
567         }
568 }
569 cease(0);
570 exit(0);
571
572