X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=3f52e39dfe8057d4fb202f364be81eb36567ceb3;hb=2b58ccdf81685a1167a43c38705a0d84b9d8d661;hp=c730773aff702f9c1cbb4d79cad00932dd2f5036;hpb=2f1b948ea733e0ece1909a31987dc8f03044e851;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index c730773a..3f52e39d 100644 --- a/perl/Msg.pm +++ b/perl/Msg.pm @@ -11,14 +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 Errno qw(EWOULDBLOCK EAGAIN EINPROGRESS); -use POSIX qw(F_GETFL F_SETFL O_NONBLOCK); -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 = (); @@ -29,6 +34,35 @@ $er_handles = IO::Select->new(); $now = time; +BEGIN { + # Checks if blocking is supported + eval { + require POSIX; POSIX->import(qw(O_NONBLOCK F_SETFL F_GETFL)) + }; + 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)); + }; +} + +my $w = $^W; +$^W = 0; +my $eagain = eval {EAGAIN()}; +my $einprogress = eval {EINPROGRESS()}; +my $ewouldblock = eval {EWOULDBLOCK()}; +$^W = $w; +$cnum = 0; + + # #----------------------------------------------------------------- # Generalised initializer @@ -47,10 +81,13 @@ sub new lineend => "\r\n", 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; } @@ -71,6 +108,8 @@ sub set_rproc sub blocking { + return unless $blocking_supported; + my $flags = fcntl ($_[0], F_GETFL, 0); if ($_[1]) { $flags &= ~O_NONBLOCK; @@ -90,10 +129,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}; } @@ -106,9 +146,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; } } @@ -134,11 +174,12 @@ sub connect { $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef; blocking($sock, 0); + $conn->{blocking} = 0; + my $ip = gethostbyname($to_host); - my $r = $sock->connect($to_port, $ip); - unless ($r) { - return undef unless $! == EINPROGRESS; - } +# my $r = $sock->connect($to_port, $ip); + my $r = connect($sock, pack_sockaddr_in($to_port, $ip)); + return undef unless $r || _err_will_block($!); $conn->{sock} = $sock; @@ -165,11 +206,9 @@ 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'); - set_event_handler ($sock, read => undef, write => undef, error => undef); - - unless ($^O =~ /^MS/i) { + unless ($main::is_win) { kill 'TERM', $conn->{pid} if exists $conn->{pid}; } @@ -181,6 +220,7 @@ sub disconnect { } return unless defined($sock); + set_event_handler ($sock, read => undef, write => undef, error => undef); shutdown($sock, 3); close($sock); } @@ -216,7 +256,10 @@ sub _send { # return to the event loop only after every message, or if it # is likely to block in the middle of a message. - blocking($sock, $flush); + if ($conn->{blocking} != $flush) { + blocking($sock, $flush); + $conn->{blocking} = $flush; + } my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0; while (@$rq) { @@ -241,20 +284,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}; @@ -264,8 +308,25 @@ sub _send { 1; # Success } +sub dup_sock +{ + my $conn = shift; + my $oldsock = $conn->{sock}; + my $rc = $rd_callbacks{$oldsock}; + my $wc = $wt_callbacks{$oldsock}; + my $ec = $er_callbacks{$oldsock}; + my $sock = $oldsock->new_from_fd($oldsock, "w+"); + if ($sock) { + set_event_handler($oldsock, read=>undef, write=>undef, error=>undef); + $conn->{sock} = $sock; + set_event_handler($sock, read=>$rc, write=>$wc, error=>$ec); + $oldsock->close; + } +} + sub _err_will_block { - return ($_[0] == EAGAIN || $_[0] == EWOULDBLOCK || $_[0] == EINPROGRESS); + return 0 unless $blocking_supported; + return ($_[0] == $eagain || $_[0] == $ewouldblock || $_[0] == $einprogress); } sub close_on_empty @@ -318,11 +379,18 @@ sub _rcv { # Complement to _send return unless defined($sock); my @lines; - blocking($sock, 0); + if ($conn->{blocking}) { + blocking($sock, 0); + $conn->{blocking} = 0; + } $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); + } } } else { if (_err_will_block($!)) { @@ -337,29 +405,37 @@ 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}; + } } } sub new_client { my $server_conn = shift; my $sock = $server_conn->{sock}->accept(); - my $conn = $server_conn->new($server_conn->{rproc}); - $conn->{sock} = $sock; - my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $sock->peerhost(), $conn->{peerport} = $sock->peerport()); - $conn->{sort} = 'Incoming'; - if ($eproc) { - $conn->{eproc} = $eproc; - set_event_handler ($sock, error => $eproc); + if ($sock) { + my $conn = $server_conn->new($server_conn->{rproc}); + $conn->{sock} = $sock; + blocking($sock, 0); + $conn->{blocking} = 0; + my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $sock->peerhost(), $conn->{peerport} = $sock->peerport()); + $conn->{sort} = 'Incoming'; + if ($eproc) { + $conn->{eproc} = $eproc; + set_event_handler ($sock, error => $eproc); + } + if ($rproc) { + $conn->{rproc} = $rproc; + my $callback = sub {$conn->_rcv}; + set_event_handler ($sock, read => $callback); + } else { # Login failed + &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc}; + $conn->disconnect(); + } + } else { + dbg("Msg: error on accept ($!)") if isdbg('err'); } - if ($rproc) { - $conn->{rproc} = $rproc; - my $callback = sub {$conn->_rcv}; - set_event_handler ($sock, read => $callback); - } else { # Login failed - &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc}; - $conn->disconnect(); - } } sub close_server @@ -372,11 +448,19 @@ 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 @@ -424,8 +508,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); - $now = time; + ($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}; @@ -445,11 +528,22 @@ sub event_loop { } } +sub sleep +{ + my ($pkg, $interval) = @_; + my $now = time; + while (time - $now < $interval) { + $pkg->event_loop(10, 0.01); + } +} + 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--; }