X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=cf15ff76a8ce2242762d01248339d0ea1c61e22e;hb=403e6bac61ed5d509aeda5b49557481ac88bc08c;hp=8be2fc1bf071d39dd464907de8397da7fdd27891;hpb=8cdf937edf493134e69b2df1ec689839148f2d8f;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index 8be2fc1b..cf15ff76 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,0)); +$main::build += $VERSION; +$main::branch += $BRANCH; + use IO::Select; use IO::Socket; use DXDebug; @@ -30,6 +37,7 @@ $now = time; BEGIN { # Checks if blocking is supported eval { + local $^W; require POSIX; POSIX->import(qw(O_NONBLOCK F_SETFL F_GETFL)) }; if ($@ || $main::is_win) { @@ -43,8 +51,34 @@ BEGIN { # import as many of these errno values as are available eval { + local $^W; require Errno; Errno->import(qw(EAGAIN EINPROGRESS EWOULDBLOCK)); }; + + unless ($^O eq 'MSWin32') { + if ($] >= 5.6) { + eval { + local $^W; + 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; @@ -80,7 +114,7 @@ sub new $noconns++; - dbg('connll', "Connection created ($noconns)"); + dbg("Connection created ($noconns)") if isdbg('connll'); return bless $conn, $class; } @@ -102,14 +136,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 +165,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}; } @@ -169,7 +212,10 @@ sub connect { blocking($sock, 0); $conn->{blocking} = 0; + # does the host resolve? my $ip = gethostbyname($to_host); + return undef unless $ip; + # my $r = $sock->connect($to_port, $ip); my $r = connect($sock, pack_sockaddr_in($to_port, $ip)); return undef unless $r || _err_will_block($!); @@ -199,12 +245,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 +254,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 { @@ -337,16 +385,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; + + unless ($main::is_win) { + 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"); + } + + eval {setsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE, 1)} or dbg("setsockopt keepalive: $!"); + eval {setsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER, pack("ll", 0, 0))} or dbg("setsockopt linger: $!"); + eval {setsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY, 1)} or eval {setsockopt($conn->{sock}, SOL_SOCKET, TCP_NODELAY, 1)} or dbg("setsockopt tcp_nodelay: $!"); + $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; @@ -379,11 +454,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($!)) { @@ -411,6 +504,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 +521,7 @@ sub new_client { $conn->disconnect(); } } else { - dbg('err', "Msg: error on accept ($!)"); + dbg("Msg: error on accept ($!)") if isdbg('err'); } } @@ -494,24 +588,35 @@ sub set_event_handler { } sub event_loop { - my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once + my ($pkg, $loop_count, $timeout, $wronly) = @_; # event_loop(1) to process events once my ($conn, $r, $w, $e, $rset, $wset, $eset); while (1) { # Quit the loop if no handles left to process - last unless ($rd_handles->count() || $wt_handles->count()); + if ($wronly) { + last unless $wt_handles->count(); - ($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}; - } - foreach $r (@$rset) { - &{$rd_callbacks{$r}}($r) if exists $rd_callbacks{$r}; - } - foreach $w (@$wset) { - &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w}; - } + ($rset, $wset, $eset) = IO::Select->select(undef, $wt_handles, undef, $timeout); + + foreach $w (@$wset) { + &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w}; + } + } else { + + last unless ($rd_handles->count() || $wt_handles->count()); + + ($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}; + } + foreach $r (@$rset) { + &{$rd_callbacks{$r}}($r) if exists $rd_callbacks{$r}; + } + foreach $w (@$wset) { + &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w}; + } + } Timer::handler; @@ -536,7 +641,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--; }