X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=f6164132a33469ae7757e171863d722146e10195;hb=b56ccffc81ac94b40755bb8c7485eb223f7e2ff4;hp=1ed5377c20f890238aba810e798f4c54ec5bfeef;hpb=4718fa3cf34f4d60c0f3e0167d9cf8ad2def44df;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index 1ed5377c..f6164132 100644 --- a/perl/Msg.pm +++ b/perl/Msg.pm @@ -11,12 +11,19 @@ package Msg; use strict; + +use vars qw($VERSION $BRANCH); +$VERSION = sprintf( "%d.%03d", q$Revision$ =~ /(\d+)\.(\d+)/ ); +$BRANCH = sprintf( "%d.%03d", q$Revision$ =~ /\d+\.\d+\.(\d+)\.(\d+)/ ) || 0; +$main::build += $VERSION; +$main::branch += $BRANCH; + use IO::Select; use IO::Socket; use DXDebug; use Timer; -use vars qw(%rd_callbacks %wt_callbacks %er_callbacks $rd_handles $wt_handles $er_handles $now %conns $noconns); +use vars qw(%rd_callbacks %wt_callbacks %er_callbacks $rd_handles $wt_handles $er_handles $now %conns $noconns $blocking_supported $cnum); %rd_callbacks = (); %wt_callbacks = (); @@ -26,19 +33,49 @@ $wt_handles = IO::Select->new(); $er_handles = IO::Select->new(); $now = time; -my $blocking_supported = 0; BEGIN { # Checks if blocking is supported eval { - require POSIX; POSIX->import(qw (F_SETFL F_GETFL O_NONBLOCK)); + require POSIX; POSIX->import(qw(O_NONBLOCK F_SETFL F_GETFL)) }; - $blocking_supported = 1 unless $@; + if ($@ || $main::is_win) { +# print STDERR "POSIX Blocking *** NOT *** supported $@\n"; + $blocking_supported = 0; + } else { + $blocking_supported = 1; +# print STDERR "POSIX Blocking enabled\n"; + } + # import as many of these errno values as are available eval { require Errno; Errno->import(qw(EAGAIN EINPROGRESS EWOULDBLOCK)); }; + + unless ($^O eq 'MSWin32') { + if ($] >= 5.6) { + eval { + require Socket; Socket->import(qw(IPPROTO_TCP TCP_NODELAY)); + }; + } else { + dbg("IPPROTO_TCP and TCP_NODELAY manually defined"); + eval 'sub IPPROTO_TCP { 6 };'; + eval 'sub TCP_NODELAY { 1 };'; + } + } + # http://support.microsoft.com/support/kb/articles/Q150/5/37.asp + # defines EINPROGRESS as 10035. We provide it here because some + # Win32 users report POSIX::EINPROGRESS is not vendor-supported. + if ($^O eq 'MSWin32') { + eval '*EINPROGRESS = sub { 10036 };'; + eval '*EWOULDBLOCK = *EAGAIN = sub { 10035 };'; + eval '*F_GETFL = sub { 0 };'; + eval '*F_SETFL = sub { 0 };'; + eval '*IPPROTO_TCP = sub { 6 };'; + eval '*TCP_NODELAY = sub { 1 };'; + $blocking_supported = 0; # it appears that this DOESN'T work :-( + } } my $w = $^W; @@ -47,6 +84,8 @@ my $eagain = eval {EAGAIN()}; my $einprogress = eval {EINPROGRESS()}; my $ewouldblock = eval {EWOULDBLOCK()}; $^W = $w; +$cnum = 0; + # #----------------------------------------------------------------- @@ -67,10 +106,12 @@ sub new csort => 'telnet', timeval => 60, blocking => 0, + cnum => (($cnum < 999) ? (++$cnum) : ($cnum = 1)), }; $noconns++; - dbg('connll', "Connection created ($noconns)"); + + dbg("Connection created ($noconns)") if isdbg('connll'); return bless $conn, $class; } @@ -92,14 +133,23 @@ sub set_rproc sub blocking { return unless $blocking_supported; - - my $flags = fcntl ($_[0], F_GETFL, 0); - if ($_[1]) { - $flags &= ~O_NONBLOCK; + + # Make the handle stop blocking, the Windows way. + if ($main::is_win) { + # 126 is FIONBIO (some docs say 0x7F << 16) + ioctl( $_[0], + 0x80000000 | (4 << 16) | (ord('f') << 8) | 126, + "$_[1]" + ); } else { - $flags |= O_NONBLOCK; + my $flags = fcntl ($_[0], F_GETFL, 0); + if ($_[1]) { + $flags &= ~O_NONBLOCK; + } else { + $flags |= O_NONBLOCK; + } + fcntl ($_[0], F_SETFL, $flags); } - fcntl ($_[0], F_SETFL, $flags); } # save it @@ -112,10 +162,11 @@ sub conns if (ref $pkg) { $call = $pkg->{call} unless $call; return undef unless $call; - confess "changing $pkg->{call} to $call" if exists $pkg->{call} && $call ne $pkg->{call}; + dbg("changing $pkg->{call} to $call") if isdbg('connll') && exists $pkg->{call} && $call ne $pkg->{call}; + delete $conns{$pkg->{call}} if exists $pkg->{call} && exists $conns{$pkg->{call}} && $pkg->{call} ne $call; $pkg->{call} = $call; $ref = $conns{$call} = $pkg; - dbg('connll', "Connection $call stored"); + dbg("Connection $pkg->{cnum} $call stored") if isdbg('connll'); } else { $ref = $conns{$call}; } @@ -128,9 +179,9 @@ sub pid_gone my ($pkg, $pid) = @_; my @pid = grep {$_->{pid} == $pid} values %conns; - for (@pid) { - &{$_->{eproc}}($_, "$pid has gorn") if exists $_->{eproc}; - $_->disconnect; + foreach my $p (@pid) { + &{$p->{eproc}}($p, "$pid has gorn") if exists $p->{eproc}; + $p->disconnect; } } @@ -155,10 +206,8 @@ sub connect { my $proto = getprotobyname('tcp'); $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef; - if ($conn->{blocking}) { - blocking($sock, 0); - $conn->{blocking} = 0; - } + blocking($sock, 0); + $conn->{blocking} = 0; my $ip = gethostbyname($to_host); # my $r = $sock->connect($to_port, $ip); @@ -190,12 +239,8 @@ sub disconnect { delete $conns{$call} if $ref && $ref == $conn; } $call ||= 'unallocated'; - dbg('connll', "Connection $call disconnected"); + dbg("Connection $conn->{cnum} $call disconnected") if isdbg('connll'); - unless ($^O =~ /^MS/i) { - kill 'TERM', $conn->{pid} if exists $conn->{pid}; - } - # get rid of any references for (keys %$conn) { if (ref($conn->{$_})) { @@ -203,10 +248,16 @@ sub disconnect { } } - return unless defined($sock); - set_event_handler ($sock, read => undef, write => undef, error => undef); - shutdown($sock, 3); - close($sock); + if (defined($sock)) { + set_event_handler ($sock, read => undef, write => undef, error => undef); + shutdown($sock, 3); + close($sock); + } + + unless ($main::is_win) { + kill 'TERM', $conn->{pid} if exists $conn->{pid}; + } + } sub send_now { @@ -268,20 +319,21 @@ sub _send { $conn->disconnect; return 0; # fail. Message remains in queue .. } - } + } elsif (isdbg('raw')) { + my $call = $conn->{call} || 'none'; + dbgdump('raw', "$call send $bytes_written: ", $msg); + } $offset += $bytes_written; $bytes_to_write -= $bytes_written; } delete $conn->{send_offset}; $offset = 0; shift @$rq; - last unless $flush; # Go back to select and wait + #last unless $flush; # Go back to select and wait # for it to fire again. } # Call me back if queue has not been drained. - if (@$rq) { - set_event_handler ($sock, write => sub {$conn->_send(0)}); - } else { + unless (@$rq) { set_event_handler ($sock, write => undef); if (exists $conn->{close_on_empty}) { &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc}; @@ -327,16 +379,43 @@ sub new_server { my $self = $pkg->new($login_proc); $self->{sock} = IO::Socket::INET->new ( - LocalAddr => $my_host, - LocalPort => $my_port, + LocalAddr => "$my_host:$my_port", +# LocalPort => $my_port, Listen => SOMAXCONN, Proto => 'tcp', - Reuse => 1); + Reuse => 1); die "Could not create socket: $! \n" unless $self->{sock}; set_event_handler ($self->{sock}, read => sub { $self->new_client } ); return $self; } + +sub nolinger +{ + my $conn = shift; + + if (isdbg('sock')) { + my ($l, $t) = unpack "ll", getsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER); + my $k = unpack 'l', getsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE); + my $n = $main::is_win ? 0 : unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY); + dbg("Linger is: $l $t, keepalive: $k, nagle: $n"); + } + + setsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER, pack("ll", 0, 0)) or confess "setsockopt linger: $!"; + setsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE, 1) or confess "setsockopt keepalive: $!"; + unless ($main::is_win) { + setsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY, 1) or confess "setsockopt: $!"; + } + $conn->{sock}->autoflush(0); + + if (isdbg('sock')) { + my ($l, $t) = unpack "ll", getsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER); + my $k = unpack 'l', getsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE); + my $n = $main::is_win ? 0 : unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY); + dbg("Linger is: $l $t, keepalive: $k, nagle: $n"); + } +} + sub dequeue { my $conn = shift; @@ -369,7 +448,29 @@ sub _rcv { # Complement to _send $bytes_read = sysread ($sock, $msg, 1024, 0); if (defined ($bytes_read)) { if ($bytes_read > 0) { - $conn->{msg} .= $msg; + if (isdbg('raw')) { + my $call = $conn->{call} || 'none'; + dbgdump('raw', "$call read $bytes_read: ", $msg); + } + if ($conn->{echo}) { + my @ch = split //, $msg; + my $out; + for (@ch) { + if (/[\cH\x7f]/) { + $out .= "\cH \cH"; + $conn->{msg} =~ s/.$//; + } else { + $out .= $_; + $conn->{msg} .= $_; + } + } + if (defined $out) { + set_event_handler ($sock, write => sub{$conn->_send(0)}); + push @{$conn->{outqueue}}, $out; + } + } else { + $conn->{msg} .= $msg; + } } } else { if (_err_will_block($!)) { @@ -384,7 +485,9 @@ FINISH: &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc}; $conn->disconnect; } else { - $conn->dequeue if exists $conn->{msg}; + unless ($conn->{disable_read}) { + $conn->dequeue if exists $conn->{msg}; + } } } @@ -394,6 +497,9 @@ sub new_client { if ($sock) { my $conn = $server_conn->new($server_conn->{rproc}); $conn->{sock} = $sock; + blocking($sock, 0); + $conn->nolinger; + $conn->{blocking} = 0; my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $sock->peerhost(), $conn->{peerport} = $sock->peerport()); $conn->{sort} = 'Incoming'; if ($eproc) { @@ -409,7 +515,7 @@ sub new_client { $conn->disconnect(); } } else { - dbg('err', "Msg: error on accept ($!)"); + dbg("Msg: error on accept ($!)") if isdbg('err'); } } @@ -423,11 +529,18 @@ sub close_server # close all clients (this is for forking really) sub close_all_clients { - for (values %conns) { - $_->disconnect; + foreach my $conn (values %conns) { + $conn->disconnect; } } +sub disable_read +{ + my $conn = shift; + set_event_handler ($conn->{sock}, read => undef); + return $_[0] ? $conn->{disable_read} = $_[0] : $_[0]; +} + # #---------------------------------------------------- # Event loop routines used by both client and server @@ -476,7 +589,7 @@ sub event_loop { # Quit the loop if no handles left to process last unless ($rd_handles->count() || $wt_handles->count()); - ($rset, $wset) = IO::Select->select($rd_handles, $wt_handles, $er_handles, $timeout); + ($rset, $wset, $eset) = IO::Select->select($rd_handles, $wt_handles, $er_handles, $timeout); foreach $e (@$eset) { &{$er_callbacks{$e}}($e) if exists $er_callbacks{$e}; @@ -509,7 +622,9 @@ sub DESTROY { my $conn = shift; my $call = $conn->{call} || 'unallocated'; - dbg('connll', "Connection $call being destroyed ($noconns)"); + my $host = $conn->{peerhost} || ''; + my $port = $conn->{peerport} || ''; + dbg("Connection $conn->{cnum} $call [$host $port] being destroyed") if isdbg('connll'); $noconns--; }