X-Git-Url: http://gb7djk.dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FMsg.pm;h=1ed5377c20f890238aba810e798f4c54ec5bfeef;hb=ee8c2e94a865274a859ffc3e57d043e24dcf39df;hp=f3dec8b1610b5743e5596b7b406bd9d9ed760304;hpb=f62f7f8ba8aa59de4f4ed8a19f0300b6804a0009;p=spider.git

diff --git a/perl/Msg.pm b/perl/Msg.pm
index f3dec8b1..1ed5377c 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,12 +155,15 @@ sub connect {
 	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;
+	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));
+	return undef unless $r || _err_will_block($!);
 	
 	$conn->{sock} = $sock;
     
@@ -167,8 +192,6 @@ 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};
 	}
@@ -181,6 +204,7 @@ sub disconnect {
 	}
 
 	return unless defined($sock);
+    set_event_handler ($sock, read => undef, write => undef, error => undef);
     shutdown($sock, 3);
 	close($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) {
@@ -344,22 +391,26 @@ FINISH:
 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;
+		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('err', "Msg: error on accept ($!)");
 	}
-    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
@@ -377,6 +428,7 @@ sub close_all_clients
 	}
 }
 
+#
 #----------------------------------------------------
 # Event loop routines used by both client and server
 
@@ -444,6 +496,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;