add more debuging,

[spider.git] / perl / Msg.pm

diff --git a/perl/Msg.pm b/perl/Msg.pm
index 403303dd05df5bafa77a2b8bb1e37c4d09e0587f..f1f60edfb6f845e5e04604a41c1a49785380a33f 100644 (file)
--- a/perl/Msg.pm
+++ b/perl/Msg.pm
@@ -15,10 +15,8 @@ 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);
+use vars qw(%rd_callbacks %wt_callbacks %er_callbacks $rd_handles $wt_handles $er_handles $now %conns $noconns $blocking_supported);
 
 %rd_callbacks = ();
 %wt_callbacks = ();
@@ -28,14 +26,25 @@ $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 O_NONBLOCK));
+        require POSIX; POSIX->import(qw(O_NONBLOCK F_SETFL F_GETFL))
     };
-    $blocking_supported = 1 unless $@;
+       if ($@ || $main::is_win) {
+               print STDERR "POSIX Blocking *** NOT *** supported $@\n";
+               $blocking_supported = 0;
+       } else {
+               $blocking_supported = 1;
+               print STDERR "POSIX Blocking enabled\n";
+       }
+
+
+       # import as many of these errno values as are available
+       eval {
+               require Errno; Errno->import(qw(EAGAIN EINPROGRESS EWOULDBLOCK));
+       };
 }
 
 my $w = $^W;
@@ -63,6 +72,7 @@ sub new
                lineend => "\r\n",
                csort => 'telnet',
                timeval => 60,
+               blocking => 0,
     };
 
        $noconns++;
@@ -152,6 +162,8 @@ sub connect {
        $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef;
        
        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));
@@ -184,7 +196,7 @@ sub disconnect {
        $call ||= 'unallocated';
        dbg('connll', "Connection $call disconnected");
        
-       unless ($^O =~ /^MS/i) {
+       unless ($main::is_win) {
                kill 'TERM', $conn->{pid} if exists $conn->{pid};
        }
 
@@ -232,7 +244,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) {
@@ -257,7 +272,10 @@ sub _send {
                                        $conn->disconnect;
                     return 0; # fail. Message remains in queue ..
                 }
-            }
+            } elsif (isdbg('raw')) {
+                               my $call = $conn->{call} || 'none';
+                               dbgdump('raw', "$call send $bytes_written: ", $msg);
+                       }
             $offset         += $bytes_written;
             $bytes_to_write -= $bytes_written;
         }
@@ -280,6 +298,22 @@ 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 unless $blocking_supported;
        return ($_[0] == $eagain || $_[0] == $ewouldblock || $_[0] == $einprogress);
@@ -335,11 +369,18 @@ 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) {
                        $conn->{msg} .= $msg;
+                       if (isdbg('raw')) {
+                               my $call = $conn->{call} || 'none';
+                               dbgdump('raw', "$call read $bytes_read: ", $msg);
+                       }
                } 
        } else {
                if (_err_will_block($!)) {
@@ -361,22 +402,28 @@ 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;
+               blocking($sock, 0);
+               $conn->{blocking} = 0;
+               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
@@ -394,6 +441,7 @@ sub close_all_clients
        }
 }
 
+#
 #----------------------------------------------------
 # Event loop routines used by both client and server
 
@@ -441,7 +489,7 @@ sub event_loop {
        # 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, $er_handles, $timeout);
+               ($rset, $wset, $eset) = IO::Select->select($rd_handles, $wt_handles, $er_handles, $timeout);
                
         foreach $e (@$eset) {
             &{$er_callbacks{$e}}($e) if exists $er_callbacks{$e};
@@ -461,6 +509,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;