X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FDXMsg.pm;h=2028f8b1da8403498950c746c2c3ab54c2408140;hb=30dbf70f84b53174005810f64f546d2181e1a8c6;hp=97aaa35514b21060cf8fc5ac78f09a47080a7a09;hpb=ca18864d1264fadfe9869b630f05c950876c2169;p=spider.git diff --git a/perl/DXMsg.pm b/perl/DXMsg.pm index 97aaa355..2028f8b1 100644 --- a/perl/DXMsg.pm +++ b/perl/DXMsg.pm @@ -26,12 +26,14 @@ use DXProtVars; use DXProtout; use DXDebug; use DXLog; -use FileHandle; +use IO::File; +use Fcntl; use Carp; use strict; use vars qw(%work @msg $msgdir %valid %busy $maxage $last_clean - @badmsg $badmsgfn $forwardfn @forward); + @badmsg $badmsgfn $forwardfn @forward $timeout $waittime + $queueinterval $lastq); %work = (); # outstanding jobs @msg = (); # messages we have @@ -40,33 +42,47 @@ $msgdir = "$main::root/msg"; # directory contain the msgs $maxage = 30 * 86400; # the maximum age that a message shall live for if not marked $last_clean = 0; # last time we did a clean @forward = (); # msg forward table +$timeout = 30*60; # forwarding timeout +$waittime = 60*60; # time an aborted outgoing message waits before trying again +$queueinterval = 5*60; # run the queue every 5 minutes +$lastq = 0; + $badmsgfn = "$msgdir/badmsg.pl"; # list of TO address we wont store $forwardfn = "$msgdir/forward.pl"; # the forwarding table %valid = ( - fromnode => '9,From Node', - tonode => '9,To Node', + fromnode => '5,From Node', + tonode => '5,To Node', to => '0,To', from => '0,From', t => '0,Msg Time,cldatetime', - private => '9,Private', + private => '5,Private', subject => '0,Subject', linesreq => '0,Lines per Gob', - rrreq => '9,Read Confirm', + rrreq => '5,Read Confirm', origin => '0,Origin', lines => '5,Data', stream => '9,Stream No', - count => '9,Gob Linecnt', - file => '9,File?,yesno', - gotit => '9,Got it Nodes,parray', - lines => '9,Lines,parray', - 'read' => '9,Times read', + count => '5,Gob Linecnt', + file => '5,File?,yesno', + gotit => '5,Got it Nodes,parray', + lines => '5,Lines,parray', + 'read' => '5,Times read', size => '0,Size', msgno => '0,Msgno', keep => '0,Keep this?,yesno', + lastt => '5,Last processed,cldatetime', + waitt => '5,Wait until,cldatetime', ); +sub DESTROY +{ + my $self = shift; + undef $self->{lines}; + undef $self->{gotit}; +} + # allocate a new object # called fromnode, tonode, from, to, datetime, private?, subject, nolinesper sub alloc @@ -87,6 +103,7 @@ sub alloc $self->{'read'} = shift; $self->{rrreq} = shift; $self->{gotit} = []; + $self->{lastt} = $main::systime; return $self; } @@ -102,16 +119,55 @@ sub workclean delete $ref->{lines}; delete $ref->{file}; delete $ref->{count}; + delete $ref->{lastt} if exists $ref->{lastt}; + delete $ref->{waitt} if exists $ref->{waitt}; } sub process { my ($self, $line) = @_; + + # this is periodic processing + if (!$self || !$line) { + + # wander down the work queue stopping any messages that have timed out + for (keys %busy) { + my $node = $_; + my $ref = $busy{$_}; + if (exists $ref->{lastt} && $main::systime > $ref->{lastt} + $timeout) { + $ref->stop_msg($node); + + # delay any outgoing messages that fail + $ref->{waitt} = $main::systime + $waittime + rand(120) if $node ne $main::mycall; + } + } + + # queue some message if the interval timer has gone off + if ($main::systime > $lastq + $queueinterval) { + queue_msg(0); + $lastq = $main::systime; + } + + # clean the message queue + clean_old() if $main::systime - $last_clean > 3600 ; + return; + } + my @f = split /\^/, $line; my ($pcno) = $f[0] =~ /^PC(\d\d)/; # just get the number - + SWITCH: { if ($pcno == 28) { # incoming message + + # first look for any messages in the busy queue + # and cancel them this should both resolve timed out incoming messages + # and crossing of message between nodes, incoming messages have priority + if (exists $busy{$f[2]}) { + my $ref = $busy{$f[2]}; + my $tonode = $ref->{tonode}; + $ref->stop_msg($self->call); + } + my $t = cltounix($f[5], $f[6]); my $stream = next_transno($f[2]); my $ref = DXMsg->alloc($stream, uc $f[3], $f[4], $t, $f[7], $f[8], $f[13], '0', $f[11]); @@ -127,6 +183,7 @@ sub process $work{"$f[2]$stream"} = $ref; # store in work $busy{$f[2]} = $ref; # set interlock $self->send(DXProt::pc30($f[2], $f[1], $stream)); # send ack + $ref->{lastt} = $main::systime; last SWITCH; } @@ -140,6 +197,7 @@ sub process dbg('msg', "stream $f[3]: $ref->{count} lines received\n"); $ref->{count} = 0; } + $ref->{lastt} = $main::systime; } last SWITCH; } @@ -157,6 +215,7 @@ sub process $ref->{lines} = []; push @{$ref->{lines}}, ($ref->read_msg_body); $ref->send_tranche($self); + $ref->{lastt} = $main::systime; } else { $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } @@ -168,6 +227,7 @@ sub process if ($ref) { dbg('msg', "tranche ack stream $f[3]\n"); $ref->send_tranche($self); + $ref->{lastt} = $main::systime; } else { $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } @@ -194,7 +254,7 @@ sub process my $m; for $m (@msg) { if ($ref->{subject} eq $m->{subject} && $ref->{t} == $m->{t} && $ref->{from} eq $m->{from}) { - $ref->stop_msg($self); + $ref->stop_msg($self->call); my $msgno = $m->{msgno}; dbg('msg', "duplicate message to $msgno\n"); Log('msg', "duplicate message to $msgno"); @@ -204,7 +264,7 @@ sub process # look for 'bad' to addresses if (grep $ref->{to} eq $_, @badmsg) { - $ref->stop_msg($self); + $ref->stop_msg($self->call); dbg('msg', "'Bad' TO address $ref->{to}"); Log('msg', "'Bad' TO address $ref->{to}"); return; @@ -215,16 +275,15 @@ sub process $ref->store($ref->{lines}); add_dir($ref); my $dxchan = DXChannel->get($ref->{to}); - $dxchan->send($dxchan->msg('msgnew')) if $dxchan; + $dxchan->send($dxchan->msg('m9')) if $dxchan && $dxchan->is_user; Log('msg', "Message $ref->{msgno} from $ref->{from} received from $f[2] for $ref->{to}"); } } - $ref->stop_msg($self); - queue_msg(0); + $ref->stop_msg($self->call); } else { $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } - queue_msg(0); + # queue_msg(0); last SWITCH; } @@ -239,10 +298,12 @@ sub process push @{$ref->{gotit}}, $f[2]; # mark this up as being received $ref->store($ref->{lines}); # re- store the file } - $ref->stop_msg($self); + $ref->stop_msg($self->call); } else { $self->send(DXProt::pc42($f[2], $f[1], $f[3])); # unknown stream } + + # send next one if present queue_msg(0); last SWITCH; } @@ -253,7 +314,7 @@ sub process $f[3] =~ s/^\///o; # remove the leading / $f[3] = lc $f[3]; # to lower case; dbg('msg', "incoming file $f[3]\n"); - last SWITCH if $f[3] =~ /^(perl|cmd|local|src|lib|include|sys|msg|connect)/; # prevent access to executables + $f[3] = 'packclus/' . $f[3] unless $f[3] =~ /^packclus\//o; # create any directories my @part = split /\//, $f[3]; @@ -286,7 +347,7 @@ sub process dbg('msg', "stream $f[3]: abort received\n"); my $ref = $work{"$f[2]$f[3]"}; if ($ref) { - $ref->stop_msg($self); + $ref->stop_msg($self->call); $ref = undef; } @@ -295,15 +356,14 @@ sub process if ($pcno == 49) { # global delete on subject for (@msg) { - if ($_->{subject} eq $f[2]) { + if ($_->{from} eq $f[1] && $_->{subject} eq $f[2]) { $_->del_msg(); - Log('msg', "Message $_->{msgno} fully deleted by $f[1]"); + Log('msg', "Message $_->{msgno} from $_->{from} ($_->{subject}) fully deleted"); + DXProt::broadcast_ak1a($line, $self); } } } } - - clean_old() if $main::systime - $last_clean > 3600 ; # clean the message queue } @@ -323,7 +383,7 @@ sub store if ($ref->{file}) { # a file dbg('msg', "To be stored in $ref->{to}\n"); - my $fh = new FileHandle "$ref->{to}", "w"; + my $fh = new IO::File "$ref->{to}", "w"; if (defined $fh) { my $line; foreach $line (@{$lines}) { @@ -343,7 +403,7 @@ sub store dbg('msg', "To be stored in $fn\n"); # now save the file, overwriting what's there, YES I KNOW OK! (I will change it if it's a problem) - my $fh = new FileHandle "$fn", "w"; + my $fh = new IO::File "$fn", "w"; if (defined $fh) { my $rr = $ref->{rrreq} ? '1' : '0'; my $priv = $ref->{private} ? '1': '0'; @@ -412,7 +472,7 @@ sub read_msg_header my @f; my $size; - $file = new FileHandle; + $file = new IO::File; if (!open($file, $fn)) { print "Error reading $fn $!\n"; return undef; @@ -457,7 +517,7 @@ sub read_msg_body my $fn = filename($msgno); my @out; - $file = new FileHandle; + $file = new IO::File; if (!open($file, $fn)) { print "Error reading $fn $!\n"; return undef; @@ -508,6 +568,13 @@ sub queue_msg foreach $ref (@msg) { # firstly, is it private and unread? if so can I find the recipient # in my cluster node list offsite? + + # ignore 'delayed' messages until their waiting time has expired + if (exists $ref->{waitt}) { + next if $ref->{waitt} < $main::systime; + delete $ref->{waitt}; + } + if ($ref->{private}) { if ($ref->{'read'} == 0) { $clref = DXCluster->get_exact($ref->{to}); @@ -570,8 +637,8 @@ sub start_msg $self->{count} = 0; $self->{tonode} = $dxchan->call; $self->{fromnode} = $main::mycall; - $busy{$dxchan->call} = $self; - $work{"$self->{tonode}"} = $self; + $busy{$self->{tonode}} = $self; + $work{$self->{tonode}} = $self; $dxchan->send(DXProt::pc28($self->{tonode}, $self->{fromnode}, $self->{to}, $self->{from}, $self->{t}, $self->{private}, $self->{subject}, $self->{origin}, $self->{rrreq})); } @@ -597,12 +664,14 @@ sub get_fwq # stop a message from continuing, clean it out, unlock interlocks etc sub stop_msg { - my ($self, $dxchan) = @_; - my $node = $dxchan->call; + my $self = shift; + my $node = shift; + my $stream = $self->{stream} if exists $self->{stream}; + - dbg('msg', "stop msg $self->{msgno} stream $self->{stream}\n"); + dbg('msg', "stop msg $self->{msgno} -> node $node\n"); delete $work{$node}; - delete $work{"$node$self->{stream}"}; + delete $work{"$node$stream"} if $stream; $self->workclean; delete $busy{$node}; } @@ -615,7 +684,7 @@ sub next_transno my $fn = "$msgdir/$name"; my $msgno; - my $fh = new FileHandle; + my $fh = new IO::File; if (sysopen($fh, $fn, O_RDWR|O_CREAT, 0666)) { $fh->autoflush(1); $msgno = $fh->getline; @@ -634,7 +703,7 @@ sub next_transno # initialise the message 'system', read in all the message headers sub init { - my $dir = new FileHandle; + my $dir = new IO::File; my @dir; my $ref; @@ -735,7 +804,7 @@ sub do_send_stuff $loc->{lines} = []; $self->state('sendbody'); #push @out, $self->msg('sendbody'); - push @out, "Enter Message /EX (^Z) to send or /ABORT (^Y) to exit"; + push @out, $self->msg('m8'); } elsif ($self->state eq 'sendbody') { confess "local var gone missing" if !ref $self->{loc}; my $loc = $self->{loc}; @@ -758,12 +827,12 @@ sub do_send_stuff $loc->{rrreq}); $ref->store($loc->{lines}); $ref->add_dir(); - #push @out, $self->msg('sendsent', $to); - push @out, "msgno $ref->{msgno} sent to $to"; + push @out, $self->msg('m11', $ref->{msgno}, $to); + #push @out, "msgno $ref->{msgno} sent to $to"; my $dxchan = DXChannel->get(uc $to); if ($dxchan) { if ($dxchan->is_user()) { - $dxchan->send("New mail has arrived for you"); + $dxchan->send($dxchan->msg('m9')); } } } @@ -772,11 +841,11 @@ sub do_send_stuff delete $loc->{to}; delete $self->{loc}; $self->func(undef); - DXMsg::queue_msg(0); + $self->state('prompt'); } elsif ($line eq "\031" || uc $line eq "/ABORT" || uc $line eq "/QUIT") { #push @out, $self->msg('sendabort'); - push @out, "aborted"; + push @out, $self->msg('m10'); delete $loc->{lines}; delete $loc->{to}; delete $self->{loc};