X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=f3dec8b1610b5743e5596b7b406bd9d9ed760304;hb=f62f7f8ba8aa59de4f4ed8a19f0300b6804a0009;hp=7af8566e09f800f56b232b471945bdd40c827581;hpb=8cf5b21b8c40d97b30a0f605247bc65210df734f;p=spider.git diff --git a/perl/Msg.pm b/perl/Msg.pm index 7af8566e..f3dec8b1 100644 --- a/perl/Msg.pm +++ b/perl/Msg.pm @@ -13,26 +13,21 @@ package Msg; use strict; use IO::Select; use IO::Socket; -#use DXDebug; +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 $rd_handles $wt_handles $now @timerchain); +use vars qw(%rd_callbacks %wt_callbacks %er_callbacks $rd_handles $wt_handles $er_handles $now %conns $noconns); %rd_callbacks = (); %wt_callbacks = (); +%er_callbacks = (); $rd_handles = IO::Select->new(); $wt_handles = IO::Select->new(); -$now = time; -@timerchain = (); - -my $blocking_supported = 0; +$er_handles = IO::Select->new(); -BEGIN { - # Checks if blocking is supported - eval { - require POSIX; POSIX->import(qw (F_SETFL O_NONBLOCK EAGAIN)); - }; - $blocking_supported = 1 unless $@; -} +$now = time; # #----------------------------------------------------------------- @@ -51,47 +46,141 @@ sub new state => 0, lineend => "\r\n", csort => 'telnet', + timeval => 60, }; + $noconns++; + dbg('connll', "Connection created ($noconns)"); return bless $conn, $class; } +sub set_error +{ + my $conn = shift; + my $callback = shift; + $conn->{eproc} = $callback; + 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 +{ + my $flags = fcntl ($_[0], F_GETFL, 0); + if ($_[1]) { + $flags &= ~O_NONBLOCK; + } else { + $flags |= O_NONBLOCK; + } + fcntl ($_[0], F_SETFL, $flags); +} + +# save it +sub conns +{ + my $pkg = shift; + my $call = shift; + my $ref; + + 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}; + $pkg->{call} = $call; + $ref = $conns{$call} = $pkg; + dbg('connll', "Connection $call stored"); + } else { + $ref = $conns{$call}; + } + return $ref; +} + +# this is only called by any dependent processes going away unexpectedly +sub pid_gone +{ + my ($pkg, $pid) = @_; + + my @pid = grep {$_->{pid} == $pid} values %conns; + for (@pid) { + &{$_->{eproc}}($_, "$pid has gorn") if exists $_->{eproc}; + $_->disconnect; + } +} + #----------------------------------------------------------------- # Send side routines sub connect { my ($pkg, $to_host, $to_port, $rproc) = @_; - # Create a new internet socket - my $sock = IO::Socket::INET->new ( - PeerAddr => $to_host, - PeerPort => $to_port, - Proto => 'tcp', - Reuse => 1); - - return undef unless $sock; - # Create a connection end-point object my $conn = $pkg; unless (ref $pkg) { $conn = $pkg->new($rproc); } + $conn->{peerhost} = $to_host; + $conn->{peerport} = $to_port; + $conn->{sort} = 'Outgoing'; + + # Create a new internet socket + my $sock = IO::Socket::INET->new(); + return undef unless $sock; + + my $proto = getprotobyname('tcp'); + $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef; + + blocking($sock, 0); + my $ip = gethostbyname($to_host); + my $r = $sock->connect($to_port, $ip); + unless ($r) { + return undef unless $! == EINPROGRESS; + } + $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; } sub disconnect { my $conn = shift; + return if exists $conn->{disconnecting}; + + $conn->{disconnecting} = 1; my $sock = delete $conn->{sock}; $conn->{state} = 'E'; - delete $conn->{cmd}; - $conn->{timeout}->del_timer if $conn->{timeout}; + $conn->{timeout}->del if $conn->{timeout}; + + # be careful to delete the correct one + my $call; + if ($call = $conn->{call}) { + my $ref = $conns{$call}; + delete $conns{$call} if $ref && $ref == $conn; + } + $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); shutdown($sock, 3); close($sock); } @@ -107,12 +196,12 @@ 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 { my $conn = shift; - push (@{$conn->{outqueue}}, $_[0]); + push (@{$conn->{outqueue}}, defined $_[0] ? $_[0] : ''); } sub _send { @@ -127,7 +216,7 @@ sub _send { # return to the event loop only after every message, or if it # is likely to block in the middle of a message. - $flush ? $conn->set_blocking() : $conn->set_non_blocking(); + blocking($sock, $flush); my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0; while (@$rq) { @@ -148,8 +237,7 @@ sub _send { # be called back eventually, and will resume sending return 1; } else { # Uh, oh - delete $conn->{send_offset}; - $conn->handle_send_err($!); + &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc}; $conn->disconnect; return 0; # fail. Message remains in queue .. } @@ -165,40 +253,25 @@ 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; + } } 1; # Success } sub _err_will_block { - if ($blocking_supported) { - return ($_[0] == EAGAIN()); - } - return 0; -} -sub set_non_blocking { # $conn->set_blocking - if ($blocking_supported) { - # preserve other fcntl flags - my $flags = fcntl ($_[0], F_GETFL(), 0); - fcntl ($_[0], F_SETFL(), $flags | O_NONBLOCK()); - } -} -sub set_blocking { - if ($blocking_supported) { - my $flags = fcntl ($_[0], F_GETFL(), 0); - $flags &= ~O_NONBLOCK(); # Clear blocking, but preserve other flags - fcntl ($_[0], F_SETFL(), $flags); - } + return ($_[0] == EAGAIN || $_[0] == EWOULDBLOCK || $_[0] == EINPROGRESS); } -sub handle_send_err { - # For more meaningful handling of send errors, subclass Msg and - # rebless $conn. - my ($conn, $err_msg) = @_; - warn "Error while sending: $err_msg \n"; - set_event_handler ($conn->{sock}, "write" => undef); +sub close_on_empty +{ + my $conn = shift; + $conn->{close_on_empty} = 1; } #----------------------------------------------------------------- @@ -212,22 +285,28 @@ sub new_server { $self->{sock} = IO::Socket::INET->new ( LocalAddr => $my_host, LocalPort => $my_port, - Listen => 5, + Listen => SOMAXCONN, 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; } sub dequeue { my $conn = shift; - my $msg; - - while ($msg = shift @{$conn->{inqueue}}){ - &{$conn->{rproc}}($conn, $msg, $!); - $! = 0; + + if ($conn->{msg} =~ /\n/) { + my @lines = split /\r?\n/, $conn->{msg}; + if ($conn->{msg} =~ /\n$/) { + delete $conn->{msg}; + } else { + $conn->{msg} = pop @lines; + } + for (@lines) { + &{$conn->{rproc}}($conn, defined $_ ? $_ : ''); + } } } @@ -239,24 +318,11 @@ sub _rcv { # Complement to _send return unless defined($sock); my @lines; - $conn->set_non_blocking(); + blocking($sock, 0); $bytes_read = sysread ($sock, $msg, 1024, 0); if (defined ($bytes_read)) { if ($bytes_read > 0) { - if ($msg =~ /\n/) { - @lines = split /\r?\n/, $msg; - $lines[0] = '' unless @lines; - $lines[0] = $conn->{msg} . $lines[0] if exists $conn->{msg}; - push @lines, ' ' unless @lines; - if ($msg =~ /\n$/) { - delete $conn->{msg}; - } else { - $conn->{msg} = pop @lines; - } - push @{$conn->{inqueue}}, @lines if @lines; - } else { - $conn->{msg} .= $msg; - } + $conn->{msg} .= $msg; } } else { if (_err_will_block($!)) { @@ -268,11 +334,10 @@ sub _rcv { # Complement to _send FINISH: if (defined $bytes_read && $bytes_read == 0) { -# $conn->disconnect(); - &{$conn->{rproc}}($conn, undef, $!); - delete $conn->{inqueue}; + &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc}; + $conn->disconnect; } else { - $conn->dequeue; + $conn->dequeue if exists $conn->{msg}; } } @@ -281,12 +346,18 @@ sub new_client { my $sock = $server_conn->{sock}->accept(); my $conn = $server_conn->new($server_conn->{rproc}); $conn->{sock} = $sock; - my $rproc = &{$server_conn->{rproc}} ($conn, $sock->peerhost(), $sock->peerport()); + 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 {_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(); } } @@ -294,10 +365,18 @@ 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; } +# close all clients (this is for forking really) +sub close_all_clients +{ + for (values %conns) { + $_->disconnect; + } +} + #---------------------------------------------------- # Event loop routines used by both client and server @@ -325,54 +404,39 @@ sub set_event_handler { $rd_handles->remove($handle); } } -} - -sub new_timer -{ - my ($pkg, $time, $proc, $recur) = @_; - my $obj = ref($pkg); - my $class = $obj || $pkg; - my $self = bless { t=>$time + time, proc=>$proc }, $class; - $self->{interval} = $time if $recur; - push @timerchain, $self; - return $self; -} - -sub del_timer -{ - my $self = shift; - @timerchain = grep {$_ != $self} @timerchain; + if (exists $args{'error'}) { + $callback = $args{'error'}; + if ($callback) { + $er_callbacks{$handle} = $callback; + $er_handles->add($handle); + } else { + delete $er_callbacks{$handle}; + $er_handles->remove($handle); + } + } } sub event_loop { my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once - my ($conn, $r, $w, $rset, $wset); + 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()); - ($rset, $wset) = - IO::Select->select ($rd_handles, $wt_handles, undef, $timeout); - $now = time; + ($rset, $wset) = 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}; + &{$rd_callbacks{$r}}($r) if exists $rd_callbacks{$r}; } foreach $w (@$wset) { &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w}; } - # handle things on the timer chain - for (@timerchain) { - if ($now >= $_->{t}) { - &{$_->{proc}}(); - $_->{t} = $now + $_->{interval} if exists $_->{interval}; - } - } - - # remove dead timers - @timerchain = grep { $_->{t} > $now } @timerchain; + Timer::handler; if (defined($loop_count)) { last unless --$loop_count; @@ -380,6 +444,14 @@ sub event_loop { } } +sub DESTROY +{ + my $conn = shift; + my $call = $conn->{call} || 'unallocated'; + dbg('connll', "Connection $call being destroyed ($noconns)"); + $noconns--; +} + 1; __END__