X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=f925681841c0746e44b7e7ee0d65b31d11a9c20c;hb=07b08c6f048732431ae45b34463387f17d68b823;hp=449f1790ca8dcb1b0156e06c31d402784264320a;hpb=586cbb347e7639f5575b48572e75140501a109c0;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index 449f1790..f9256818 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; # #----------------------------------------------------------------- @@ -62,8 +81,17 @@ sub set_error set_event_handler($conn->{sock}, error => $callback) if exists $conn->{sock}; } +sub set_rproc +{ + my $conn = shift; + my $callback = shift; + $conn->{rproc} = $callback; +} + sub blocking { + return unless $blocking_supported; + my $flags = fcntl ($_[0], F_GETFL, 0); if ($_[1]) { $flags &= ~O_NONBLOCK; @@ -128,16 +156,15 @@ sub connect { blocking($sock, 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; if ($conn->{rproc}) { - my $callback = sub {_rcv($conn)}; - set_event_handler ($sock, "read" => $callback); + my $callback = sub {$conn->_rcv}; + set_event_handler ($sock, read => $callback); } return $conn; } @@ -149,9 +176,6 @@ sub disconnect { $conn->{disconnecting} = 1; my $sock = delete $conn->{sock}; $conn->{state} = 'E'; - delete $conn->{cmd}; - delete $conn->{eproc}; - delete $conn->{rproc}; $conn->{timeout}->del if $conn->{timeout}; # be careful to delete the correct one @@ -163,11 +187,19 @@ sub disconnect { $call ||= 'unallocated'; dbg('connll', "Connection $call disconnected"); - set_event_handler ($sock, read => undef, write => undef, error => undef); unless ($^O =~ /^MS/i) { kill 'TERM', $conn->{pid} if exists $conn->{pid}; } + + # get rid of any references + for (keys %$conn) { + if (ref($conn->{$_})) { + delete $conn->{$_}; + } + } + return unless defined($sock); + set_event_handler ($sock, read => undef, write => undef, error => undef); shutdown($sock, 3); close($sock); } @@ -183,7 +215,7 @@ sub send_later { $conn->enqueue($msg); my $sock = $conn->{sock}; return unless defined($sock); - set_event_handler ($sock, "write" => sub {$conn->_send(0)}); + set_event_handler ($sock, write => sub {$conn->_send(0)}); } sub enqueue { @@ -240,9 +272,9 @@ sub _send { } # Call me back if queue has not been drained. if (@$rq) { - set_event_handler ($sock, "write" => sub {$conn->_send(0)}); + set_event_handler ($sock, write => sub {$conn->_send(0)}); } else { - set_event_handler ($sock, "write" => undef); + set_event_handler ($sock, write => undef); if (exists $conn->{close_on_empty}) { &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc}; $conn->disconnect; @@ -252,7 +284,8 @@ sub _send { } 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 @@ -276,7 +309,7 @@ sub new_server { Proto => 'tcp', Reuse => 1); die "Could not create socket: $! \n" unless $self->{sock}; - set_event_handler ($self->{sock}, "read" => sub { $self->new_client } ); + set_event_handler ($self->{sock}, read => sub { $self->new_client } ); return $self; } @@ -321,8 +354,8 @@ sub _rcv { # Complement to _send FINISH: if (defined $bytes_read && $bytes_read == 0) { - &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc}; - $conn->disconnect(); + &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc}; + $conn->disconnect; } else { $conn->dequeue if exists $conn->{msg}; } @@ -337,12 +370,12 @@ sub new_client { $conn->{sort} = 'Incoming'; if ($eproc) { $conn->{eproc} = $eproc; - set_event_handler ($sock, "error" => $eproc); + set_event_handler ($sock, error => $eproc); } if ($rproc) { $conn->{rproc} = $rproc; - my $callback = sub {_rcv($conn)}; - set_event_handler ($sock, "read" => $callback); + my $callback = sub {$conn->_rcv}; + set_event_handler ($sock, read => $callback); } else { # Login failed &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc}; $conn->disconnect(); @@ -352,7 +385,7 @@ sub new_client { sub close_server { my $conn = shift; - set_event_handler ($conn->{sock}, "read" => undef); + set_event_handler ($conn->{sock}, read => undef, write => undef, error => undef ); $conn->{sock}->close; } @@ -412,7 +445,6 @@ sub event_loop { last unless ($rd_handles->count() || $wt_handles->count()); ($rset, $wset) = IO::Select->select($rd_handles, $wt_handles, $er_handles, $timeout); - $now = time; foreach $e (@$eset) { &{$er_callbacks{$e}}($e) if exists $er_callbacks{$e};