X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=165596d270265b34c2fc5cba785cc633f1f58b5c;hb=532c5c9e14a48ec17d38dce5d904a3b612a2563b;hp=3b9dbd36f2a4f546f4b25169f1360b0552bed40d;hpb=95345d68a5a8ac618021c0786c7234258b903f6e;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index 3b9dbd36..165596d2 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 $blocking_supported); +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 = (); @@ -33,11 +40,11 @@ BEGIN { require POSIX; POSIX->import(qw(O_NONBLOCK F_SETFL F_GETFL)) }; if ($@ || $main::is_win) { - print STDERR "POSIX Blocking *** NOT *** supported $@\n"; +# print STDERR "POSIX Blocking *** NOT *** supported $@\n"; $blocking_supported = 0; } else { $blocking_supported = 1; - print STDERR "POSIX Blocking enabled\n"; +# print STDERR "POSIX Blocking enabled\n"; } @@ -45,6 +52,27 @@ BEGIN { eval { require Errno; Errno->import(qw(EAGAIN EINPROGRESS EWOULDBLOCK)); }; + + eval { + require Socket; Socket->import(qw(IPPROTO_TCP TCP_NODELAY)); + }; + unless (*IPPROTO_TCP && !$^O =~ /^MS/) { + dbg("IPPROTO_TCP and TCP_NODELAY manually defined"); + eval '*IPPROTO_TCP = sub { 6 };'; + eval '*TCP_NODELAY = sub { 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 = 1; + } } my $w = $^W; @@ -53,6 +81,8 @@ my $eagain = eval {EAGAIN()}; my $einprogress = eval {EINPROGRESS()}; my $ewouldblock = eval {EWOULDBLOCK()}; $^W = $w; +$cnum = 0; + # #----------------------------------------------------------------- @@ -73,10 +103,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; } @@ -98,14 +130,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 @@ -118,10 +159,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}; } @@ -134,9 +176,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; } } @@ -194,12 +236,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 ($main::is_win) { - kill 'TERM', $conn->{pid} if exists $conn->{pid}; - } - # get rid of any references for (keys %$conn) { if (ref($conn->{$_})) { @@ -207,10 +245,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 { @@ -272,20 +316,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}; @@ -331,16 +376,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 = 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 = unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY); + dbg("Linger is: $l $t, keepalive: $k, nagle: $n"); + } +} + sub dequeue { my $conn = shift; @@ -373,7 +445,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($!)) { @@ -388,7 +482,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}; + } } } @@ -399,6 +495,7 @@ sub new_client { 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'; @@ -415,7 +512,7 @@ sub new_client { $conn->disconnect(); } } else { - dbg('err', "Msg: error on accept ($!)"); + dbg("Msg: error on accept ($!)") if isdbg('err'); } } @@ -429,11 +526,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 @@ -482,7 +586,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}; @@ -515,7 +619,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--; }