X-Git-Url: http://dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=13833962d3d316d26f997e8861a431d143bda478;hb=309eea6bcd4a0eaf25f810168bc38dc348c2a982;hp=e3385d9166585436dad106bd94e8b94aa6b0c411;hpb=abbcfa7500858a2eba4135b0af5db9f3fca8d68e;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index e3385d91..13833962 100644 --- a/perl/Msg.pm +++ b/perl/Msg.pm @@ -20,15 +20,16 @@ use Mojo::IOLoop::Stream; use DXDebug; use Timer; -use vars qw($now %conns $noconns $cnum $total_in $total_out $connect_timeout $disc_waittime); +use vars qw($now %conns $noconns $cnum $total_in $total_out $total_lines_in $total_lines_out $connect_timeout $disc_waittime); $total_in = $total_out = 0; +$total_lines_in = $total_lines_out = 0; $now = time; $cnum = 0; $connect_timeout = 5; -$disc_waittime = 3; +$disc_waittime = 1.5; our %delqueue; @@ -43,15 +44,19 @@ sub new my $class = $obj || $pkg; my $conn = { - rproc => $rproc, - inqueue => [], - outqueue => [], - state => 0, - lineend => "\r\n", - csort => 'telnet', - timeval => 60, - blocking => 0, - cnum => (($cnum < 999) ? (++$cnum) : ($cnum = 1)), + rproc => $rproc, + inqueue => [], + outqueue => [], + state => 0, + lineend => "\r\n", + csort => 'telnet', + timeval => 60, + blocking => 0, + cnum => (($cnum < 999) ? (++$cnum) : ($cnum = 1)), + linesin => 0, + linesout => 0, + datain => 0, + dataout => 0, }; $noconns++; @@ -146,7 +151,7 @@ sub _on_connect $sock->timeout(0); $sock->start; $conn->{peerhost} = eval { $handle->peerhost; }; - dbg((ref $conn) . " connected $conn->{cnum} to $conn->{peerhost}:$conn->{peerport}") if isdbg('connll'); + dbg((ref $conn) . " connected $conn->{cnum} to $conn->{peerhost}:$conn->{peerport}") if isdbg('conn') || isdbg ('connect'); if ($conn->{on_connect}) { &{$conn->{on_connect}}($conn, $handle); } @@ -177,9 +182,18 @@ sub connect { my $sock; $conn->{sock} = $sock = Mojo::IOLoop::Client->new; - $sock->on(connect => sub {$conn->_on_connect($_[1])} ); - $sock->on(error => sub {&{$conn->{eproc}}($conn, $_[1]) if exists $conn->{eproc}; $conn->disconnect}); - $sock->on(close => sub {$conn->disconnect}); + $sock->on(connect => sub { + $conn->_on_connect($_[1]) + } ); + $sock->on(error => sub { + &{$conn->{eproc}}($conn, $_[1]) if exists $conn->{eproc}; + delete $conn->{sock}; + $conn->disconnect + }); + $sock->on(close => sub { + delete $conn->{sock}; + $conn->disconnect} + ); # copy any args like on_connect, on_disconnect etc while (my ($k, $v) = each %args) { @@ -244,41 +258,41 @@ sub disconnect { my $conn = shift; my $count = $conn->{disconnecting}++; - if (isdbg('connll')) { - my ($pkg, $fn, $line) = caller; - dbg((ref $conn) . "::disconnect on call $conn->{call} attempt $conn->{disconnecting} called from ${pkg}::${fn} line $line "); + my $dbg = isdbg('connll'); + my ($pkg, $fn, $line) = caller if $dbg; + + if ($count >= 2) { + dbgtrace((ref $conn) . "::disconnect on call $conn->{call} attempt $conn->{disconnecting} called from ${pkg}::${fn} line $line FORCING CLOSE ") if $dbg; + _close_it($conn); + return; } + dbg((ref $conn) . "::disconnect on call $conn->{call} attempt $conn->{disconnecting} called from ${pkg}::${fn} line $line ") if $dbg; return if $count; - + # remove this conn from the active queue + # be careful to delete the correct one + my $call; + if ($call = $conn->{call}) { + my $ref = $conns{$call}; + delete $conns{$call} if $ref && $ref == $conn; + } + $call ||= 'unallocated'; + + $delqueue{$conn} = $conn; # save this connection until everything is finished my $sock = $conn->{sock}; if ($sock) { - - # remove me from the active list - my $call; - if ($call = $conn->{call}) { - my $ref = $conns{$call}; - delete $conns{$call} if $ref && $ref == $conn; + if ($sock->{buffer}) { + my $lth = length $sock->{buffer}; + Mojo::IOLoop->timer($disc_waittime, sub { + dbg("Buffer contained $lth characters, coordinated for $disc_waittime secs, now disconnecting $call") if $dbg; + _close_it($conn); + }); + } else { + dbg("Buffer empty, just close $call") if $dbg; + _close_it($conn); } - $conn->{delay} = Mojo::IOLoop->delay ( -# Mojo::IOLoop->delay ( - sub { - my $delay = shift; - dbg("before drain $call"); - $sock->on(drain => $delay->begin); - 1; - }, - sub { - my $delay = shift; - _close_it($conn); - 1; - } - ); - $conn->{delay}->wait; - - $delqueue{$conn} = $conn; # save this connection until everything is finished } else { - dbg((ref $conn) . " socket missing on $conn->{call}") if isdbg('connll'); + dbg((ref $conn) . " socket missing on $conn->{call}") if $dbg; _close_it($conn); } } @@ -290,18 +304,13 @@ sub _close_it $conn->{state} = 'E'; $conn->{timeout}->del if $conn->{timeout}; + my $call = $conn->{call}; + if (isdbg('connll')) { my ($pkg, $fn, $line) = caller; dbg((ref $conn) . "::_close_it on call $conn->{call} attempt $conn->{disconnecting} called from ${pkg}::${fn} line $line "); } - # be careful to delete the correct one - my $call; - if ($call = $conn->{call}) { - my $ref = $conns{$call}; - delete $conns{$call} if $ref && $ref == $conn; - } - $call ||= 'unallocated'; dbg((ref $conn) . " Connection $conn->{cnum} $call starting to close") if isdbg('connll'); @@ -311,7 +320,7 @@ sub _close_it if ($sock) { dbg((ref $conn) . " Connection $conn->{cnum} $call closing gracefully") if isdbg('connll'); - $sock->close_gracefully; + $sock->close_gracefully if $sock->can('close_gracefully'); } # get rid of any references @@ -341,13 +350,14 @@ sub _send_stuff my $lth = length $data; my $call = $conn->{call} || 'none'; if (isdbg('raw')) { - if (isdbg('raw')) { - dbgdump('raw', "$call send $lth: ", $lth); - } + dbgdump('raw', "$call send $lth:", $data); } if (defined $sock) { $sock->write($data); $total_out += $lth; + $conn->{dataout} += $lth; + ++$conn->{linesout}; + ++$total_lines_out; } else { dbg("_send_stuff $call ending data ignored: $data"); } @@ -423,6 +433,8 @@ sub dequeue } else { $conn->{msg} = pop @lines; } + $conn->{linesin} += @lines; + $total_lines_in += @lines; for (@lines) { last if $conn->{disconnecting}; &{$conn->{rproc}}($conn, defined $_ ? $_ : ''); @@ -438,8 +450,8 @@ sub _rcv { # Complement to _send return if $conn->{disconnecting}; $total_in += length $msg; + $conn->{datain} += length $msg; - my @lines; if (isdbg('raw')) { my $call = $conn->{call} || 'none'; my $lth = length $msg; @@ -478,9 +490,11 @@ sub new_client { $sock->on(read => sub {$conn->_rcv($_[1])}); $sock->timeout(0); $sock->start; - dbg((ref $conn) . "accept $conn->{cnum} from $conn->{peerhost} $conn->{peerport}") if isdbg('connll'); - - my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $handle->peerhost, $conn->{peerport} = $handle->peerport); + $conn->{peerhost} = $handle->peerhost || 'unknown'; + $conn->{peerhost} =~ s|^::ffff:||; # chop off leading pseudo IPV6 stuff on dual stack listeners + $conn->{peerport} = $handle->peerport || 0; + dbg((ref $conn) . " accept $conn->{cnum} from $conn->{peerhost}:$conn->{peerport}") if isdbg('conn') || isdbg('connect'); + my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost}, $conn->{peerport}); $conn->{sort} = 'Incoming'; if ($eproc) { $conn->{eproc} = $eproc; @@ -547,8 +561,7 @@ sub DESTROY if (isdbg('connll')) { my ($pkg, $fn, $line) = caller; - dbg((ref $conn) . "::DESTROY on call $call called from ${pkg}::${fn} line $line "); - + dbgtrace((ref $conn) . "::DESTROY on call $call called from ${pkg}::${fn} line $line "); } my $call = $conn->{call} || 'unallocated'; @@ -557,7 +570,8 @@ sub DESTROY my $sock = $conn->{sock}; if ($sock) { - $sock->close_gracefully; + $sock->close_gracefully if $sock->can('close_gracefully'); + delete $conn->{sock}; } $noconns--;