]> dxcluster.org Git - spider.git/commitdiff
do non blocking connects
authorminima <minima>
Tue, 13 Mar 2001 14:22:13 +0000 (14:22 +0000)
committerminima <minima>
Tue, 13 Mar 2001 14:22:13 +0000 (14:22 +0000)
fix memory leak in connects

14 files changed:
Changes
cmd/Aliases
cmd/bye.pl
cmd/disconnect.pl
cmd/stat/channel.pl
perl/DXChannel.pm
perl/DXCommandmode.pm
perl/DXProt.pm
perl/ExtMsg.pm
perl/IntMsg.pm
perl/Msg.pm
perl/Timer.pm [new file with mode: 0644]
perl/cluster.pl
perl/console.pl

diff --git a/Changes b/Changes
index e5828220a8efae422f6cca02c7fe654a58c277ac..549e49d806d27ef19d956c996ddabfd0b3152644 100644 (file)
--- a/Changes
+++ b/Changes
@@ -1,3 +1,6 @@
+13Mar01=======================================================================
+1. implemented first cut at non blocking connect
+2. removed memory leakage in connects
 10Mar01=======================================================================
 1. minor changes to the admin manual to reflect differences in distibutions
 thanks to pa3ezl (g0vgs)
index c6aab03c56c0f7c46f8cbe2a5d6720bb84fee37d..b916742cbef7ff373e47db0cee4587c138c4fc06 100644 (file)
@@ -65,6 +65,7 @@ package CmdAlias;
          '^l$', 'directory', 'directory',
          '^ll$', 'directory', 'directory',
          '^ll/(\d+)', 'directory $1', 'directory',
+         '^lm$', 'directory own', 'directory',
        ],
        'm' => [
        ],
index 1fc73f66648a9251cb327d9c5923902adc20d308..d6e2d14becc76790295e429a02a093bb15653f54 100644 (file)
@@ -5,5 +5,15 @@
 #
 
 my $self = shift;
+
+# log out text
+if ($self->is_user && -e "$main::data/logout") {
+       open(I, "$main::data/logout") or confess;
+       my @in = <I>;
+       close(I);
+       $self->send_now('D', @in);
+       sleep(1);
+}
+
 $self->state('bye');
 return (1);
index 9207d73b930c5833725821538524e1ac51c242c2..9e6b67dff927a521d5cb7db530927cf6aec8ef4d 100644 (file)
@@ -23,7 +23,7 @@ foreach $call (@calls) {
                } 
                $dxchan->disconnect;
                push @out, $self->msg('disc2', $call);
-       } elsif (my $conn = Msg->call($call)) {
+       } elsif (my $conn = Msg->conns($call)) {
                $conn->disconnect;
        } else {
                push @out, $self->msg('e10', $call);
index 147c150af6bd307e71261e4ef547aae2df306f98..8b20e0cbd759148ee63d465dc2e7ba55dd5c9d49 100644 (file)
@@ -15,7 +15,7 @@ foreach $call (@list) {
   $call = uc $call;
   my $ref = DXChannel->get($call);
   if ($ref) {
-    @out = print_all_fields($self, $ref, "Channe Information $call");
+    @out = print_all_fields($self, $ref, "Channel Information $call");
   } else {
     return (0, "Channel: $call not found") if !$ref;
   }
index b7b2d4bb84ec21ae1d76c5a3bd2f9ba89993fc97..3e32f429cfe39e745bae74f3f3283ce160a342d4 100644 (file)
@@ -370,12 +370,11 @@ sub disconnect
 {
        my $self = shift;
        my $user = $self->{user};
-       my $conn = $self->{conn};
        my $call = $self->{call};
        
-       $self->finish($conn);
+       $self->finish;
        $user->close() if defined $user;
-       $conn->disconnect() if $conn;
+       $self->{conn}->disconnect;
        $self->del();
 }
 
index 3c45aee968f0c1c0ebc8c1c560f8f0e184d51d5a..aea2064e615b3754bfb6ec849428211455702d94 100644 (file)
@@ -387,7 +387,6 @@ sub process
 sub finish
 {
        my $self = shift;
-       my $conn = shift;
        my $call = $self->call;
 
        # reset the redirection of messages back to 'normal' if we are the sysop
@@ -399,20 +398,6 @@ sub finish
        # I was the last node visited
     $self->user->node($main::mycall);
                
-       # log out text
-       if ($conn && -e "$main::data/logout") {
-               open(I, "$main::data/logout") or confess;
-               my @in = <I>;
-               close(I);
-               $self->send_now('D', @in);
-               sleep(1);
-       }
-
-#      if ($call eq $main::myalias) { # unset the channel if it is us really
-#              my $node = DXNode->get($main::mycall);
-#              $node->{dxchan} = 0;
-#      }
-       
        # issue a pc17 to everybody interested
        my $nchan = DXChannel->get($main::mycall);
        my $pc17 = $nchan->pc17($self);
index 516779b67458791266a7d00bd25453d30732b1ad..fd47a28a2860afa87f393bc2bc4e309793076272 100644 (file)
@@ -1148,7 +1148,6 @@ sub finish
 {
        my $self = shift;
        my $call = $self->call;
-       my $conn = shift;
        my $ref = DXCluster->get_exact($call);
        
        # unbusy and stop and outgoing mail
index cd18eb9318039974b7ae7f95101c7c9aeb534f46..8772955350f78c72381c37a3d2b4dcf95210c74e 100644 (file)
@@ -57,47 +57,56 @@ sub dequeue
 {
        my $conn = shift;
        my $msg;
-       
-       while (@{$conn->{inqueue}}){
-               $msg = shift @{$conn->{inqueue}};
-               dbg('connect', $msg) unless $conn->{state} eq 'C';
-               
-               $msg =~ s/\xff\xfa.*\xff\xf0|\xff[\xf0-\xfe].//g; # remove telnet options
-               $msg =~ s/[\x00-\x08\x0a-\x1f\x80-\x9f]/./g;         # immutable CSI sequence + control characters
 
-               if ($conn->{state} eq 'C') {
-                       &{$conn->{rproc}}($conn, "I$conn->{call}|$msg", $!);
-                       $! = 0;
-               } elsif ($conn->{state} eq 'WL' ) {
-                       $msg = uc $msg;
-                       if (is_callsign($msg)) {
-                               &{$conn->{rproc}}($conn, "A$msg|telnet");
-                               _send_file($conn, "$main::data/connected");
-                               $conn->{state} = 'C';
-                       } else {
-                               $conn->send_now("Sorry $msg is an invalid callsign");
-                               $conn->disconnect;
-                       }
-               } elsif ($conn->{state} eq 'WC') {
-                       if (exists $conn->{cmd} && @{$conn->{cmd}}) {
-                               $conn->_docmd($msg);
-                               if ($conn->{state} eq 'WC' && exists $conn->{cmd} &&  @{$conn->{cmd}} == 0) {
-                                       $conn->{state} = 'C';
-                                       &{$conn->{rproc}}($conn, "O$conn->{call}|telnet");
-                                       delete $conn->{cmd};
-                                       $conn->{timeout}->del_timer if $conn->{timeout};
-                               }
-                       }
+       if ($conn->{state} eq 'WC') {
+               if (exists $conn->{cmd}) {
+                       if (@{$conn->{cmd}}) {
+                               dbg('connect', $conn->{msg});
+                               $conn->_docmd($conn->{msg});
+                       } 
                }
-       }
-       if ($conn->{msg} && $conn->{state} eq 'WC' && exists $conn->{cmd} && @{$conn->{cmd}}) {
-               dbg('connect', $conn->{msg});
-               $conn->_docmd($conn->{msg});
                if ($conn->{state} eq 'WC' && exists $conn->{cmd} && @{$conn->{cmd}} == 0) {
                        $conn->{state} = 'C';
                        &{$conn->{rproc}}($conn, "O$conn->{call}|telnet");
                        delete $conn->{cmd};
-                       $conn->{timeout}->del_timer if $conn->{timeout};
+                       $conn->{timeout}->del if $conn->{timeout};
+               }
+       } elsif ($conn->{msg} =~ /\n/) {
+               my @lines = split /\r?\n/, $conn->{msg};
+               if ($conn->{msg} =~ /\n$/) {
+                       delete $conn->{msg};
+               } else {
+                       $conn->{msg} = pop @lines;
+               }
+               while (defined ($msg = shift @lines)) {
+                       dbg('connect', $msg) unless $conn->{state} eq 'C';
+               
+                       $msg =~ s/\xff\xfa.*\xff\xf0|\xff[\xf0-\xfe].//g; # remove telnet options
+                       $msg =~ s/[\x00-\x08\x0a-\x1f\x80-\x9f]/./g;         # immutable CSI sequence + control characters
+                       
+                       if ($conn->{state} eq 'C') {
+                               &{$conn->{rproc}}($conn, "I$conn->{call}|$msg");
+                       } elsif ($conn->{state} eq 'WL' ) {
+                               $msg = uc $msg;
+                               if (is_callsign($msg)) {
+                                       &{$conn->{rproc}}($conn, "A$msg|telnet");
+                                       _send_file($conn, "$main::data/connected");
+                                       $conn->{state} = 'C';
+                               } else {
+                                       $conn->send_now("Sorry $msg is an invalid callsign");
+                                       $conn->disconnect;
+                               }
+                       } elsif ($conn->{state} eq 'WC') {
+                               if (exists $conn->{cmd} && @{$conn->{cmd}}) {
+                                       $conn->_docmd($msg);
+                                       if ($conn->{state} eq 'WC' && exists $conn->{cmd} &&  @{$conn->{cmd}} == 0) {
+                                               $conn->{state} = 'C';
+                                               &{$conn->{rproc}}($conn, "O$conn->{call}|telnet");
+                                               delete $conn->{cmd};
+                                               $conn->{timeout}->del if $conn->{timeout};
+                                       }
+                               }
+                       }
                }
        }
 }
@@ -108,7 +117,11 @@ sub new_client {
     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());
+       if ($eproc) {
+               $conn->{eproc} = $eproc;
+        set_event_handler ($sock, "error" => $eproc);
+       }
     if ($rproc) {
         $conn->{rproc} = $rproc;
         my $callback = sub {$conn->_rcv};
@@ -209,8 +222,8 @@ sub _dotimeout
        my $conn = shift;
        my $val = shift;
        dbg('connect', "timeout set to $val");
-       $conn->{timeout}->del_timer if $conn->{timeout};
-       $conn->{timeout} = ExtMsg->new_timer($val, sub{ _timeout($conn); });
+       my $old = $conn->{timeout}->del if $conn->{timeout};
+       $conn->{timeout} = Timer->new($val, sub{ &_timeout($conn) });
        $conn->{timeval} = $val;
 }
 
@@ -269,7 +282,7 @@ sub _doclient
        $conn->{state} = 'C';
        &{$conn->{rproc}}($conn, "O$conn->{call}|telnet");
        delete $conn->{cmd};
-       $conn->{timeout}->del_timer if $conn->{timeout};
+       $conn->{timeout}->del if $conn->{timeout};
 }
 
 sub _send_file
@@ -287,5 +300,4 @@ sub _send_file
                        $f->close;
                }
        }
-       $! = undef;
 }
index 150ec91a2f0a860c9a63ad73a2f2a81c74d73a2e..8065f3024e5c693e04e7b50620a5d136e83c2a00 100644 (file)
@@ -26,12 +26,23 @@ sub enqueue
 sub dequeue
 {
        my $conn = shift;
-       my $msg;
-       
-       while ($msg = shift @{$conn->{inqueue}}){
-               $msg =~ s/\%([2-9A-F][0-9A-F])/chr(hex($1))/eg;
-               $msg =~ s/[\x00-\x08\x0a-\x1f\x80-\x9f]/./g;         # immutable CSI sequence + control characters
-               &{$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) {
+                       if (defined $_) {
+                               s/\%([0-9A-F][0-9A-F])/chr(hex($1))/eg;
+                               s/[\x00-\x08\x0a-\x1f\x80-\x9f]/./g;         # immutable CSI sequence + control characters
+                       } else {
+                               $_ = '';
+                       }
+                       &{$conn->{rproc}}($conn, $_);
+               }
        }
 }
+
index 7d5b407213ec6bad7c6b743174d9fe84afca88e1..449f1790ca8dcb1b0156e06c31d402784264320a 100644 (file)
@@ -13,26 +13,21 @@ package Msg;
 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 $now @timerchain %conns);
+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;
 
 #
 #-----------------------------------------------------------------
@@ -54,9 +49,30 @@ sub new
                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 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
 {
@@ -70,6 +86,7 @@ sub conns
                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};
        }
@@ -83,11 +100,8 @@ sub pid_gone
        
        my @pid = grep {$_->{pid} == $pid} values %conns;
        for (@pid) {
-               if ($_->{rproc}) {
-                       &{$_->{rproc}}($_, undef, "$pid has gorn");
-               } else {
-                       $_->disconnect;
-               }
+               &{$_->{eproc}}($_, "$pid has gorn") if exists $_->{eproc};
+               $_->disconnect;
        }
 }
 
@@ -101,17 +115,24 @@ sub connect {
        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 (
-                                      PeerAddr => $to_host,
-                                      PeerPort => $to_port,
-                                      Proto    => 'tcp',
-                                      Reuse    => 1,
-                                                                         Timeout  => $conn->{timeval} / 2);
-
+    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}) {
@@ -123,18 +144,26 @@ sub connect {
 
 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};
+       delete $conn->{eproc};
+       delete $conn->{rproc};
+       $conn->{timeout}->del if $conn->{timeout};
 
        # be careful to delete the correct one
-       if (my $call = $conn->{call}) {
+       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);
+    set_event_handler ($sock, read => undef, write => undef, error => undef);
        unless ($^O =~ /^MS/i) {
                kill 'TERM', $conn->{pid} if exists $conn->{pid};
        }
@@ -159,7 +188,7 @@ sub send_later {
 
 sub enqueue {
     my $conn = shift;
-    push (@{$conn->{outqueue}}, $_[0]);
+    push (@{$conn->{outqueue}}, defined $_[0] ? $_[0] : '');
 }
 
 sub _send {
@@ -174,7 +203,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) {
@@ -195,8 +224,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 ..
                 }
@@ -215,37 +243,22 @@ sub _send {
         set_event_handler ($sock, "write" => sub {$conn->_send(0)});
     } else {
         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;
 }
 
 #-----------------------------------------------------------------
@@ -259,7 +272,7 @@ 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};
@@ -270,11 +283,17 @@ sub new_server {
 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 $_ ? $_ : '');
+               }
        }
 }
 
@@ -286,27 +305,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;
-                               if (@lines) {
-                                       $lines[0] = $conn->{msg} . $lines[0] if exists $conn->{msg};
-                               } else {
-                                       $lines[0] = $conn->{msg} 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($!)) {
@@ -318,11 +321,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, undef) if exists $conn->{eproc};
+               $conn->disconnect();
     } else {
-               $conn->dequeue;
+               $conn->dequeue if exists $conn->{msg};
        }
 }
 
@@ -331,12 +333,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);
     } else {  # Login failed
+               &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc};
         $conn->disconnect();
     }
 }
@@ -348,6 +356,14 @@ sub close_server
        $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
 
@@ -375,54 +391,40 @@ 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);
+               ($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};
+        }
         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;
@@ -430,6 +432,14 @@ sub event_loop {
     }
 }
 
+sub DESTROY
+{
+       my $conn = shift;
+       my $call = $conn->{call} || 'unallocated';
+       dbg('connll', "Connection $call being destroyed ($noconns)");
+       $noconns--;
+}
+
 1;
 
 __END__
diff --git a/perl/Timer.pm b/perl/Timer.pm
new file mode 100644 (file)
index 0000000..8969756
--- /dev/null
@@ -0,0 +1,49 @@
+#
+# Polled Timer handling
+#
+# This uses callbacks. BE CAREFUL!!!!
+#
+# $Id$
+#
+# Copyright (c) 2001 Dirk Koopman G1TLH
+#
+
+package Timer;
+
+use vars qw(@timerchain);
+
+@timerchain = ();
+
+sub new
+{
+    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
+{
+       my $self = shift;
+       my $old = delete $self->{proc};
+       @timerchain = grep {$_ != $self} @timerchain;
+       return $old;
+}
+
+sub handler
+{
+       my $now = time;
+       
+       # handle things on the timer chain
+       for (@timerchain) {
+               if ($now >= $_->{t}) {
+                       &{$_->{proc}}();
+                       $_->{t} = $now + $_->{interval} if exists $_->{interval};
+               }
+       }
+}
+
+1;
index 2c94dcc1cb619571baef325fe4607071c453bb1c..eef7a40cd02d4fa5c970517192c35d3aba9aa689 100755 (executable)
@@ -94,38 +94,27 @@ sub already_conn
        
        dbg('chan', "-> D $call $mess\n"); 
        $conn->send_now("D$call|$mess");
-       sleep(1);
-       dbg('chan', "-> Z $call bye\n");
-       $conn->send_now("Z$call|bye"); # this will cause 'client' to disconnect
-       sleep(1);
+       sleep(2);
        $conn->disconnect;
 }
 
+sub error_handler
+{
+       my $dxchan = shift;
+       $dxchan->disconnect;
+}
+
 # handle incoming messages
 sub rec
 {
-       my ($conn, $msg, $err) = @_;
+       my ($conn, $msg) = @_;
        my $dxchan = DXChannel->get_by_cnum($conn); # get the dxconnnect object for this message
+       my ($sort, $call, $line) = DXChannel::decode_input(0, $msg);
+       return unless defined $sort;
        
-       if (!defined $msg || (defined $err && $err)) {
-               if ($dxchan) {
-                       if (defined $err) {
-                               $conn->disconnect;
-                               undef $conn;
-                               $dxchan->conn(undef);
-                       }
-                       $dxchan->disconnect;
-               } elsif ($conn) {
-                       $conn->disconnect;
-               }
-               return;
-       }
-       
-       # set up the basic channel info - this needs a bit more thought - there is duplication here
+       # set up the basic channel info
        if (!defined $dxchan) {
-               my ($sort, $call, $line) = DXChannel::decode_input(0, $msg);
-               return unless defined $sort;
+
                # is there one already connected to me - locally? 
                my $user = DXUser->get($call);
                if ($sort ne 'O' && Msg->conns($call)) {
@@ -158,13 +147,13 @@ sub rec
                # is he locked out ?
                if ($user->lockout) {
                        Log('DXCommand', "$call is locked out, disconnected");
-                       $conn->send_now("Z$call|bye"); # this will cause 'client' to disconnect
                        $conn->disconnect;
                        return;
                }
 
                # mark him up
                $conn->conns($call) unless $sort eq 'O';
+               $conn->set_error(sub {error_handler($dxchan)});
                
                # create the channel
                $dxchan = DXCommandmode->new($call, $conn, $user) if $user->is_user;
@@ -335,10 +324,14 @@ DXUser->init($userfn, 1);
 use Listeners;
 
 dbg('err', "starting listeners ...");
-push @listeners, IntMsg->new_server("$clusteraddr", $clusterport, \&login);
+my $conn = IntMsg->new_server($clusteraddr, $clusterport, \&login);
+$conn->conns("Server $clusteraddr/$clusterport");
+push @listeners, $conn;
 dbg('err', "Internal port: $clusteraddr $clusterport");
 for (@main::listen) {
-       push @listeners, ExtMsg->new_server($_->[0], $_->[1], \&login);
+       $conn = ExtMsg->new_server($_->[0], $_->[1], \&login);
+       $conn->conns("Server $_->[0]/$_->[1]");
+       push @listeners, $conn;
        dbg('err', "External Port: $_->[0] $_->[1]");
 }
 
@@ -422,7 +415,7 @@ dbg('err', "orft we jolly well go ...");
 for (;;) {
 #      $DB::trace = 1;
        
-       Msg->event_loop(1, 0.1);
+       Msg->event_loop(10, 0.001);
        my $timenow = time;
        process_inqueue();                      # read in lines from the input queue and despatch them
 #      $DB::trace = 0;
index 208eb90a4b8969fd8fbd8e95d4386fdd07f39f85..aee1bc9ba485253b28b67eba16885602473fc25d 100755 (executable)
@@ -104,9 +104,7 @@ sub do_resize
 sub cease
 {
        my $sendz = shift;
-#      if ($conn && $sendz) {
-#              $conn->send_now("Z$call|bye...");
-#      }
+       $conn->disconnect if $conn;
        endwin();
        dbgclose();
        print @_ if @_;
@@ -437,6 +435,8 @@ if (! $conn) {
        exit(0);
 }
 
+$conn->set_error(sub{cease(0)});
+
 
 unless ($DB::VERSION) {
        $SIG{'INT'} = \&sig_term;