X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=5aef961ee4f4162593228d0c1abd71a41dd52fef;hb=60bd15a823797c01182ebfb8b6b3a5bf10065b42;hp=3b96c812e502b51b16f5b6a493cbae14aa3e4822;hpb=6374fb31da62987e676244c38eb6cf73c81bf891;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index 3b96c812..5aef961e 100644 --- a/perl/Msg.pm +++ b/perl/Msg.pm @@ -11,6 +11,13 @@ 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; @@ -45,6 +52,28 @@ BEGIN { 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 };'; + $blocking_supported = 1; + } } my $w = $^W; @@ -80,7 +109,7 @@ sub new $noconns++; - dbg('connll', "Connection created ($noconns)"); + dbg("Connection created ($noconns)") if isdbg('connll'); return bless $conn, $class; } @@ -102,14 +131,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 @@ -122,11 +160,11 @@ sub conns if (ref $pkg) { $call = $pkg->{call} unless $call; return undef unless $call; - dbg('connll', "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 $pkg->{cnum} $call stored"); + dbg("Connection $pkg->{cnum} $call stored") if isdbg('connll'); } else { $ref = $conns{$call}; } @@ -199,12 +237,8 @@ sub disconnect { delete $conns{$call} if $ref && $ref == $conn; } $call ||= 'unallocated'; - dbg('connll', "Connection $conn->{cnum} $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->{$_})) { @@ -212,10 +246,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 { @@ -287,13 +327,11 @@ sub _send { 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}; @@ -339,16 +377,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; @@ -381,11 +446,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($!)) { @@ -400,7 +483,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}; + } } } @@ -411,6 +496,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'; @@ -427,7 +513,7 @@ sub new_client { $conn->disconnect(); } } else { - dbg('err', "Msg: error on accept ($!)"); + dbg("Msg: error on accept ($!)") if isdbg('err'); } } @@ -446,6 +532,13 @@ sub close_all_clients } } +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 @@ -529,7 +622,7 @@ sub DESTROY my $call = $conn->{call} || 'unallocated'; my $host = $conn->{peerhost} || ''; my $port = $conn->{peerport} || ''; - dbg('connll', "Connection $conn->{cnum} $call [$host $port] being destroyed"); + dbg("Connection $conn->{cnum} $call [$host $port] being destroyed") if isdbg('connll'); $noconns--; }