X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=e167269363d6f674ac24a016ce2fbabea51a7c0b;hb=c644e2f01b7528fb3cd4666a552a5ee282462e88;hp=839b5e453313e05db16f6f536ba26d3e4b1cdd31;hpb=2bbb601161d60a41289bd19b73fc35c24d5c6b71;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index 839b5e45..e1672693 100644 --- a/perl/Msg.pm +++ b/perl/Msg.pm @@ -15,8 +15,6 @@ 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); @@ -28,6 +26,27 @@ $wt_handles = IO::Select->new(); $er_handles = IO::Select->new(); $now = time; +my $blocking_supported = 0; + +BEGIN { + # Checks if blocking is supported + eval { + require POSIX; POSIX->import(qw (F_SETFL F_GETFL O_NONBLOCK)); + }; + $blocking_supported = 1 unless $@; + + # 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; # #----------------------------------------------------------------- @@ -47,6 +66,7 @@ sub new lineend => "\r\n", csort => 'telnet', timeval => 60, + blocking => 0, }; $noconns++; @@ -71,6 +91,8 @@ sub set_rproc sub blocking { + return unless $blocking_supported; + my $flags = fcntl ($_[0], F_GETFL, 0); if ($_[1]) { $flags &= ~O_NONBLOCK; @@ -133,13 +155,15 @@ sub connect { my $proto = getprotobyname('tcp'); $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef; - blocking($sock, 0); + if ($conn->{blocking}) { + blocking($sock, 0); + $conn->{blocking} = 0; + } + my $ip = gethostbyname($to_host); # my $r = $sock->connect($to_port, $ip); my $r = connect($sock, pack_sockaddr_in($to_port, $ip)); - unless ($r) { - return undef unless $! == EINPROGRESS; - } + return undef unless $r || _err_will_block($!); $conn->{sock} = $sock; @@ -216,7 +240,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) { @@ -264,8 +291,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,7 +362,10 @@ 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) { @@ -377,6 +424,7 @@ sub close_all_clients } } +# #---------------------------------------------------- # Event loop routines used by both client and server @@ -444,6 +492,15 @@ 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;