X-Git-Url: http://dxcluster.org/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=839b5e453313e05db16f6f536ba26d3e4b1cdd31;hb=2bbb601161d60a41289bd19b73fc35c24d5c6b71;hp=b3816df3f324ade51480e1f65d4105cc13d13b6b;hpb=3c0c1870d27827056e62012828bc7233abde1f01;p=spider.git

diff --git a/perl/Msg.pm b/perl/Msg.pm
index b3816df3..839b5e45 100644
--- a/perl/Msg.pm
+++ b/perl/Msg.pm
@@ -10,94 +10,205 @@
 
 package Msg;
 
-require Exporter;
-@ISA = qw(Exporter);
-
 use strict;
 use IO::Select;
 use IO::Socket;
-use Carp;
+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);
+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();
-my $blocking_supported = 0;
+$er_handles   = IO::Select->new();
+
+$now = time;
+
+#
+#-----------------------------------------------------------------
+# Generalised initializer
 
-BEGIN {
-    # Checks if blocking is supported
-    eval {
-        require POSIX; POSIX->import(qw (F_SETFL O_NONBLOCK EAGAIN));
+sub new
+{
+    my ($pkg, $rproc) = @_;
+	my $obj = ref($pkg);
+	my $class = $obj || $pkg;
+
+    my $conn = {
+        rproc => $rproc,
+		inqueue => [],
+		outqueue => [],
+		state => 0,
+		lineend => "\r\n",
+		csort => 'telnet',
+		timeval => 60,
     };
-    $blocking_supported = 1 unless $@;
+
+	$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,$rcvd_notification_proc) = @_;
-    
-    # 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;
+    my ($pkg, $to_host, $to_port, $rproc) = @_;
 
     # Create a connection end-point object
-    my $conn = {
-        sock                   => $sock,
-        rcvd_notification_proc => $rcvd_notification_proc,
-    };
+    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);
+	my $r = connect($sock, pack_sockaddr_in($to_port, $ip));
+	unless ($r) {
+		return undef unless $! == EINPROGRESS;
+	}
+	
+	$conn->{sock} = $sock;
     
-    if ($rcvd_notification_proc) {
-        my $callback = sub {_rcv($conn)};
-        set_event_handler ($sock, "read" => $callback);
+    if ($conn->{rproc}) {
+        my $callback = sub {$conn->_rcv};
+        set_event_handler ($sock, read => $callback);
     }
-    return bless $conn, $pkg;
+    return $conn;
 }
 
 sub disconnect {
     my $conn = shift;
+	return if exists $conn->{disconnecting};
+
+	$conn->{disconnecting} = 1;
     my $sock = delete $conn->{sock};
-    return unless defined($sock);
-    set_event_handler ($sock, "read" => undef, "write" => undef);
+	$conn->{state} = 'E';
+	$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");
+	
+	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);
 }
 
 sub send_now {
     my ($conn, $msg) = @_;
-    _enqueue ($conn, $msg);
+    $conn->enqueue($msg);
     $conn->_send (1); # 1 ==> flush
 }
 
 sub send_later {
     my ($conn, $msg) = @_;
-    _enqueue($conn, $msg);
+    $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, $msg) = @_;
-    # prepend length (encoded as network long)
-    my $len = length($msg);
-	$msg =~ s/([\%\x00-\x1f\x7f-\xff])/sprintf("%%%02X", ord($1))/eg; 
-    push (@{$conn->{queue}}, $msg . "\n");
+sub enqueue {
+    my $conn = shift;
+    push (@{$conn->{outqueue}}, defined $_[0] ? $_[0] : '');
 }
 
 sub _send {
     my ($conn, $flush) = @_;
     my $sock = $conn->{sock};
     return unless defined($sock);
-    my ($rq) = $conn->{queue};
+    my $rq = $conn->{outqueue};
 
     # If $flush is set, set the socket to blocking, and send all
     # messages in the queue - return only if there's an error
@@ -105,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) {
@@ -126,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 ..
                 }
@@ -143,60 +253,61 @@ 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;
 }
 
 #-----------------------------------------------------------------
 # Receive side routines
 
-my ($g_login_proc,$g_pkg);
-my $main_socket = 0;
 sub new_server {
-    @_ == 4 || die "Msg->new_server (myhost, myport, login_proc)\n";
+    @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
     my ($pkg, $my_host, $my_port, $login_proc) = @_;
-    
-    $main_socket = IO::Socket::INET->new (
+	my $self = $pkg->new($login_proc);
+	
+    $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 $main_socket;
-    set_event_handler ($main_socket, "read" => \&_new_client);
-    $g_login_proc = $login_proc; $g_pkg = $pkg;
+    die "Could not create socket: $! \n" unless $self->{sock};
+    set_event_handler ($self->{sock}, read => sub { $self->new_client }  );
+	return $self;
+}
+
+sub dequeue
+{
+	my $conn = shift;
+
+	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 $_ ? $_ : '');
+		}
+	}
 }
 
 sub _rcv {                     # Complement to _send
@@ -207,21 +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 /\n/, $msg;
-				$lines[0] = $conn->{msg} . $lines[0] if $conn->{msg};
-				if ($msg =~ /\n$/) {
-					delete $conn->{msg};
-				} else {
-					$conn->{msg} = pop @lines;
-				}
-			} else {
-				$conn->{msg} .= $msg;
-			}
+			$conn->{msg} .= $msg;
 		} 
 	} else {
 		if (_err_will_block($!)) {
@@ -233,41 +334,47 @@ sub _rcv {                     # Complement to _send
 
 FINISH:
     if (defined $bytes_read && $bytes_read == 0) {
-#		$conn->disconnect();
-		&{$conn->{rcvd_notification_proc}}($conn, undef, $!);
-		@lines = ();
-    } 
-
-	while (@lines){
-		$msg = shift @lines;
-		$msg =~ s/%([0-9A-Fa-f]{2})/chr(hex($1))/eg;
-		&{$conn->{rcvd_notification_proc}}($conn, $msg, $!);
-		$! = 0;
+		&{$conn->{eproc}}($conn, $!) if exists $conn->{eproc};
+		$conn->disconnect;
+    } else {
+		$conn->dequeue if exists $conn->{msg};
 	}
 }
 
-sub _new_client {
-    my $sock = $main_socket->accept();
-    my $conn = bless {
-        'sock' =>  $sock,
-        'state' => 'connected'
-    }, $g_pkg;
-    my $rcvd_notification_proc =
-        &$g_login_proc ($conn, $sock->peerhost(), $sock->peerport());
-    if ($rcvd_notification_proc) {
-        $conn->{rcvd_notification_proc} = $rcvd_notification_proc;
-        my $callback = sub {_rcv($conn)};
-        set_event_handler ($sock, "read" => $callback);
+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 ($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
 {
-	set_event_handler ($main_socket, "read" => undef);
-	$main_socket->close;
-	$main_socket = 0;
+	my $conn = shift;
+	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;
+	}
 }
 
 #----------------------------------------------------
@@ -297,28 +404,54 @@ sub set_event_handler {
             $rd_handles->remove($handle);
        }
     }
+    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
+ 
+       # 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);
+        
+		($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};
         }
+
+		Timer::handler;
+		
         if (defined($loop_count)) {
             last unless --$loop_count;
         }
     }
 }
 
+sub DESTROY
+{
+	my $conn = shift;
+	my $call = $conn->{call} || 'unallocated';
+	dbg('connll', "Connection $call being destroyed ($noconns)");
+	$noconns--;
+}
+
 1;
 
 __END__