X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=cf15ff76a8ce2242762d01248339d0ea1c61e22e;hb=403e6bac61ed5d509aeda5b49557481ac88bc08c;hp=03151ca1976aa8906a2b7d7566143ae65dd83f8f;hpb=61660841afb3901002602e4956f09de5567bc950;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index 03151ca1..cf15ff76 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,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 = (); @@ -30,21 +37,48 @@ $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) { - 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"; } # 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; @@ -53,6 +87,8 @@ my $eagain = eval {EAGAIN()}; my $einprogress = eval {EINPROGRESS()}; my $ewouldblock = eval {EWOULDBLOCK()}; $^W = $w; +$cnum = 0; + # #----------------------------------------------------------------- @@ -73,10 +109,12 @@ sub new csort => 'telnet', timeval => 60, blocking => 0, - cnum => ++$noconns, + cnum => (($cnum < 999) ? (++$cnum) : ($cnum = 1)), }; - dbg('connll', "Connection created ($noconns)"); + $noconns++; + + dbg("Connection created ($noconns)") if isdbg('connll'); return bless $conn, $class; } @@ -98,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 @@ -118,10 +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}; } @@ -164,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($!); @@ -194,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->{$_})) { @@ -207,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 { @@ -282,13 +335,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}; @@ -334,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; @@ -376,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($!)) { @@ -395,7 +491,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}; + } } } @@ -406,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'; @@ -422,7 +521,7 @@ sub new_client { $conn->disconnect(); } } else { - dbg('err', "Msg: error on accept ($!)"); + dbg("Msg: error on accept ($!)") if isdbg('err'); } } @@ -441,6 +540,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 @@ -482,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; @@ -524,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--; }