From 38aa5af305ed46fcf56f4624eb6bac29afaccda5 Mon Sep 17 00:00:00 2001 From: Dirk Koopman Date: Thu, 22 Dec 2011 20:41:45 +0000 Subject: [PATCH] WIP AnyEvent version --- perl/Console.pm | 3 + perl/ExtMsg.pm | 75 +++---- perl/Msg.pm | 584 ++++++++++++++---------------------------------- perl/Version.pm | 6 +- perl/cluster.pl | 8 +- perl/console.pl | 150 +++++++------ 6 files changed, 295 insertions(+), 531 deletions(-) diff --git a/perl/Console.pm b/perl/Console.pm index a6dc6613..4dc72d42 100644 --- a/perl/Console.pm +++ b/perl/Console.pm @@ -31,6 +31,9 @@ package main; +use vars qw($maxkhist $maxshist $foreground $background $mycallcolor @colors ); +use Curses; + $maxkhist = 100; $maxshist = 500; if ($ENV{'TERM'} =~ /(xterm|ansi)/) { diff --git a/perl/ExtMsg.pm b/perl/ExtMsg.pm index 6b2cce7d..7b5bb672 100644 --- a/perl/ExtMsg.pm +++ b/perl/ExtMsg.pm @@ -54,11 +54,8 @@ sub enqueue sub send_raw { my ($conn, $msg) = @_; - my $sock = $conn->{sock}; - return unless defined($sock); - push (@{$conn->{outqueue}}, $msg); dbg("connect $conn->{cnum}: $msg") if $conn->{state} ne 'C' && isdbg('connect'); - Msg::set_event_handler ($sock, "write" => sub {$conn->_send(0)}); + $conn->SUPER::send_raw($msg); } sub echo @@ -155,8 +152,7 @@ sub to_connected delete $conn->{timeout}; $conn->{csort} = $sort; unless ($conn->ax25) { - eval {$conn->{peerhost} = $conn->{sock}->peerhost}; - $conn->nolinger; + eval {$conn->{peerhost} = $conn->{sock}->peername}; } &{$conn->{rproc}}($conn, "$dir$call|$sort"); $conn->_send_file("$main::data/connected") unless $conn->{outgoing}; @@ -164,43 +160,44 @@ sub to_connected sub new_client { my $server_conn = shift; - my $sock = $server_conn->{sock}->accept(); + my $sock = shift; + my $peerhost = shift; + my $peerport = shift; if ($sock) { my $conn = $server_conn->new($server_conn->{rproc}); - $conn->{sock} = $sock; - $conn->nolinger; - Msg::blocking($sock, 0); + $conn->{sock} = AnyEvent::Handle->new( + + fh => $sock, + + on_eof => sub {$conn->disconnect}, + + on_error => sub {$conn->disconnect}, + + keepalive => 1, + + linger => 0, + ); $conn->{blocking} = 0; - eval {$conn->{peerhost} = $sock->peerhost}; - if ($@) { - dbg($@) if isdbg('connll'); - $conn->disconnect; + my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $peerhost, $conn->{peerport} = $peerport); + dbg("accept $conn->{cnum} from $conn->{peerhost} $conn->{peerport}") if isdbg('connll'); + $conn->{sock}->on_read(sub{$conn->_rcv}); + if ($eproc) { + $conn->{eproc} = $eproc; + } + if ($rproc) { + $conn->{rproc} = $rproc; + # send login prompt + $conn->{state} = 'WL'; + # $conn->send_raw("\xff\xfe\x01\xff\xfc\x01\ff\fd\x22"); + # $conn->send_raw("\xff\xfa\x22\x01\x01\xff\xf0"); + # $conn->send_raw("\xFF\xFC\x01"); + $conn->_send_file("$main::data/issue"); + $conn->send_raw("login: "); + $conn->_dotimeout(60); + $conn->{echo} = 1; } else { - eval {$conn->{peerport} = $sock->peerport}; - $conn->{peerport} = 0 if $@; - my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost}, $conn->{peerport}); - dbg("accept $conn->{cnum} from $conn->{peerhost} $conn->{peerport}") if isdbg('connll'); - if ($eproc) { - $conn->{eproc} = $eproc; - Msg::set_event_handler ($sock, "error" => $eproc); - } - if ($rproc) { - $conn->{rproc} = $rproc; - my $callback = sub {$conn->_rcv}; - Msg::set_event_handler ($sock, "read" => $callback); - # send login prompt - $conn->{state} = 'WL'; - # $conn->send_raw("\xff\xfe\x01\xff\xfc\x01\ff\fd\x22"); - # $conn->send_raw("\xff\xfa\x22\x01\x01\xff\xf0"); - # $conn->send_raw("\xFF\xFC\x01"); - $conn->_send_file("$main::data/issue"); - $conn->send_raw("login: "); - $conn->_dotimeout(60); - $conn->{echo} = 1; - } else { - &{$conn->{eproc}}() if $conn->{eproc}; - $conn->disconnect(); - } + &{$conn->{eproc}}() if $conn->{eproc}; + $conn->disconnect(); } } else { dbg("ExtMsg: error on accept ($!)") if isdbg('err'); diff --git a/perl/Msg.pm b/perl/Msg.pm index 5aebab65..ea2747dc 100644 --- a/perl/Msg.pm +++ b/perl/Msg.pm @@ -14,90 +14,18 @@ use strict; use DXUtil; +use AnyEvent; +use AnyEvent::Handle; +use AnyEvent::Socket; + use DXDebug; use Timer; -use vars qw(%rd_callbacks %wt_callbacks %er_callbacks $rd_handles $wt_handles $er_handles $now %conns $noconns $blocking_supported $cnum $total_in $total_out $io_socket); +use vars qw(%conns $noconns $cnum $total_in $total_out); -%rd_callbacks = (); -%wt_callbacks = (); -%er_callbacks = (); -$rd_handles = IO::Select->new(); -$wt_handles = IO::Select->new(); -$er_handles = IO::Select->new(); $total_in = $total_out = 0; - -$now = time; - -BEGIN { - # Checks if blocking is supported - eval { - local $^W; - require POSIX; POSIX->import(qw(O_NONBLOCK F_SETFL F_GETFL)) - }; - - eval { - local $^W; - require IO::Socket::INET6; - }; - - if ($@) { - dbg($@); - require IO::Socket; - $io_socket = 'IO::Socket::INET'; - } else { - $io_socket = 'IO::Socket::INET6'; - } - $io_socket->import; - - if ($@ || $main::is_win) { - $blocking_supported = $io_socket->can('blocking') ? 2 : 0; - } else { - $blocking_supported = $io_socket->can('blocking') ? 2 : 1; - } - - - # import as many of these errno values as are available - eval { - local $^W; - require Errno; Errno->import(qw(EAGAIN EINPROGRESS EWOULDBLOCK)); - }; - - unless ($^O eq 'MSWin32') { - if ($] >= 5.6) { - eval { - local $^W; - require Socket; Socket->import(qw(IPPROTO_TCP TCP_NODELAY)); - }; - } else { - dbg("IPPROTO_TCP and TCP_NODELAY manually defined"); - eval 'sub IPPROTO_TCP { 6 };'; - eval 'sub TCP_NODELAY { 1 };'; - } - } - # http://support.microsoft.com/support/kb/articles/Q150/5/37.asp - # defines EINPROGRESS as 10035. We provide it here because some - # Win32 users report POSIX::EINPROGRESS is not vendor-supported. - if ($^O eq 'MSWin32') { - eval '*EINPROGRESS = sub { 10036 };' unless defined *EINPROGRESS; - eval '*EWOULDBLOCK = *EAGAIN = sub { 10035 };' unless defined *EWOULDBLOCK; - eval '*F_GETFL = sub { 0 };' unless defined *F_GETFL; - eval '*F_SETFL = sub { 0 };' unless defined *F_SETFL; - eval 'sub IPPROTO_TCP { 6 };'; - eval 'sub TCP_NODELAY { 1 };'; - $blocking_supported = 0; # it appears that this DOESN'T work :-( - } -} - -my $w = $^W; -$^W = 0; -my $eagain = eval {EAGAIN()}; -my $einprogress = eval {EINPROGRESS()}; -my $ewouldblock = eval {EWOULDBLOCK()}; -$^W = $w; $cnum = 0; - # #----------------------------------------------------------------- # Generalised initializer @@ -131,7 +59,6 @@ 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 @@ -141,24 +68,6 @@ sub set_rproc $conn->{rproc} = $callback; } -sub blocking -{ - return unless $blocking_supported; - - # Make the handle stop blocking, the Windows way. - if ($blocking_supported) { - $_[0]->blocking($_[1]); - } else { - 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 { @@ -214,42 +123,29 @@ sub connect { # Create a connection end-point object my $conn = $pkg; - unless (ref $pkg) { - $conn = $pkg->new($rproc); - } + $conn = $pkg->new($rproc); $conn->{peerhost} = $to_host; $conn->{peerport} = $to_port; $conn->{sort} = 'Outgoing'; - my $sock; - if ($blocking_supported) { - $sock = $io_socket->new(PeerAddr => $to_host, PeerPort => $to_port, Proto => 'tcp', Blocking =>0) or return undef; - } else { - # Create a new internet socket - $sock = $io_socket->new(); - return undef unless $sock; + my $sock = AnyEvent::Handle->new( - my $proto = getprotobyname('tcp'); - $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef; + connect => [$to_host, $to_port], - blocking($sock, 0); - $conn->{blocking} = 0; +# on_connect => sub {my $h = shift; $conn->{peerhost} = $h->handle->peername;}, - # does the host resolve? - my $ip = gethostbyname($to_host); - return undef unless $ip; + on_eof => sub {$conn->disconnect}, - my $r = connect($sock, pack_sockaddr_in($to_port, $ip)); - return undef unless $r || _err_will_block($!); - } + on_error => sub {$conn->disconnect}, + + keepalive => 1, + + linger => 0, + ); $conn->{sock} = $sock; - $conn->{peerhost} = $sock->peerhost; # for consistency + $sock->on_read(sub{$conn->_rcv}); - if ($conn->{rproc}) { - my $callback = sub {$conn->_rcv}; - set_event_handler ($sock, read => $callback); - } return $conn; } @@ -258,47 +154,47 @@ sub start_program my ($conn, $line, $sort) = @_; my $pid; - local $^F = 10000; # make sure it ain't closed on exec - my ($a, $b) = $io_socket->socketpair(AF_UNIX, SOCK_STREAM, PF_UNSPEC); - if ($a && $b) { - $a->autoflush(1); - $b->autoflush(1); - $pid = fork; - if (defined $pid) { - if ($pid) { - close $b; - $conn->{sock} = $a; - $conn->{csort} = $sort; - $conn->{lineend} = "\cM" if $sort eq 'ax25'; - $conn->{pid} = $pid; - if ($conn->{rproc}) { - my $callback = sub {$conn->_rcv}; - Msg::set_event_handler ($a, read => $callback); - } - dbg("connect $conn->{cnum}: started pid: $conn->{pid} as $line") if isdbg('connect'); - } else { - $^W = 0; - dbgclose(); - STDIN->close; - STDOUT->close; - STDOUT->close; - *STDIN = IO::File->new_from_fd($b, 'r') or die; - *STDOUT = IO::File->new_from_fd($b, 'w') or die; - *STDERR = IO::File->new_from_fd($b, 'w') or die; - close $a; - unless ($main::is_win) { - # $SIG{HUP} = 'IGNORE'; - $SIG{HUP} = $SIG{CHLD} = $SIG{TERM} = $SIG{INT} = 'DEFAULT'; - alarm(0); - } - exec "$line" or dbg("exec '$line' failed $!"); - } - } else { - dbg("cannot fork for $line"); - } - } else { - dbg("no socket pair $! for $line"); - } +# local $^F = 10000; # make sure it ain't closed on exec +# my ($a, $b) = $io_socket->socketpair(AF_UNIX, SOCK_STREAM, PF_UNSPEC); +# if ($a && $b) { +# $a->autoflush(1); +# $b->autoflush(1); +# $pid = fork; +# if (defined $pid) { +# if ($pid) { +# close $b; +# $conn->{sock} = $a; +# $conn->{csort} = $sort; +# $conn->{lineend} = "\cM" if $sort eq 'ax25'; +# $conn->{pid} = $pid; +# if ($conn->{rproc}) { +# my $callback = sub {$conn->_rcv}; +# Msg::set_event_handler ($a, read => $callback); +# } +# dbg("connect $conn->{cnum}: started pid: $conn->{pid} as $line") if isdbg('connect'); +# } else { +# $^W = 0; +# dbgclose(); +# STDIN->close; +# STDOUT->close; +# STDOUT->close; +# *STDIN = IO::File->new_from_fd($b, 'r') or die; +# *STDOUT = IO::File->new_from_fd($b, 'w') or die; +# *STDERR = IO::File->new_from_fd($b, 'w') or die; +# close $a; +# unless ($main::is_win) { +# # $SIG{HUP} = 'IGNORE'; +# $SIG{HUP} = $SIG{CHLD} = $SIG{TERM} = $SIG{INT} = 'DEFAULT'; +# alarm(0); +# } +# exec "$line" or dbg("exec '$line' failed $!"); +# } +# } else { +# dbg("cannot fork for $line"); +# } +# } else { +# dbg("no socket pair $! for $line"); +# } return $pid; } @@ -329,9 +225,7 @@ sub disconnect } if (defined($sock)) { - set_event_handler ($sock, read => undef, write => undef, error => undef); - shutdown($sock, 2); - close($sock); + $sock->destroy; } unless ($main::is_win) { @@ -339,115 +233,59 @@ sub disconnect } } -sub send_now { - my ($conn, $msg) = @_; - $conn->enqueue($msg); - $conn->_send (1); # 1 ==> flush +sub _send_stuff +{ + my $conn = shift; + my $rq = $conn->{outqueue}; + my $sock = $conn->{sock}; + + while (@$rq) { + my $data = shift @$rq; + my $lth = length $data; + if (isdbg('raw')) { + my $call = $conn->{call} || 'none'; + if (isdbg('raw')) { + dbgdump('raw', "$call send $lth: ", $lth); + } + } + $sock->push_write($data); + $total_out = $lth; + } } sub send_later { my ($conn, $msg) = @_; - $conn->enqueue($msg); - my $sock = $conn->{sock}; - return unless defined($sock); - set_event_handler ($sock, write => sub {$conn->_send(0)}); -} + my $rq = $conn->{outqueue}; + my $sock = $conn->{sock}; -sub enqueue { - my $conn = shift; - push (@{$conn->{outqueue}}, defined $_[0] ? $_[0] : ''); + # this is done like this because enqueueing may be going on independently of + # sending (whether later or now) + $conn->enqueue($msg); + _send_stuff($conn) } -sub _send { - my ($conn, $flush) = @_; - my $sock = $conn->{sock}; - return unless defined($sock); - 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 - # If $flush is 0 (deferred mode) make the socket non-blocking, and - # return to the event loop only after every message, or if it - # is likely to block in the middle of a message. +sub send_now { goto &send_later; } -# if ($conn->{blocking} != $flush) { -# blocking($sock, $flush); -# $conn->{blocking} = $flush; -# } - my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0; - - while (@$rq) { - my $msg = $rq->[0]; - my $mlth = length($msg); - my $bytes_to_write = $mlth - $offset; - my $bytes_written = 0; - confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0; - while ($bytes_to_write > 0) { - $bytes_written = syswrite ($sock, $msg, - $bytes_to_write, $offset); - if (!defined($bytes_written)) { - if (_err_will_block($!)) { - # Should happen only in deferred mode. Record how - # much we have already sent. - $conn->{send_offset} = $offset; - # Event handler should already be set, so we will - # be called back eventually, and will resume sending - return 1; - } else { # Uh, oh - &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc}; - $conn->disconnect; - return 0; # fail. Message remains in queue .. - } - } elsif (isdbg('raw')) { - my $call = $conn->{call} || 'none'; - dbgdump('raw', "$call send $bytes_written: ", $msg); - } - $total_out += $bytes_written; - $offset += $bytes_written; - $bytes_to_write -= $bytes_written; - } - delete $conn->{send_offset}; - $offset = 0; - shift @$rq; - #last unless $flush; # Go back to select and wait - # for it to fire again. - } - # Call me back if queue has not been drained. - unless (@$rq) { - 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 send_raw +{ + my ($conn, $msg) = @_; + push @{$conn->{outqueue}}, $msg; + _send_stuff($conn); } -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 enqueue { + my $conn = shift; + push (@{$conn->{outqueue}}, defined $_[0] ? $_[0] : ''); } sub _err_will_block { - return 0 unless $blocking_supported; - return ($_[0] == $eagain || $_[0] == $ewouldblock || $_[0] == $einprogress); + return 0; } sub close_on_empty { my $conn = shift; - $conn->{close_on_empty} = 1; + $conn->{sock}->push_shutdown; } #----------------------------------------------------------------- @@ -458,14 +296,8 @@ sub new_server { my ($pkg, $my_host, $my_port, $login_proc) = @_; my $self = $pkg->new($login_proc); - $self->{sock} = $io_socket->new ( - LocalAddr => "$my_host:$my_port", -# LocalPort => $my_port, - Listen => SOMAXCONN, - Proto => 'tcp', - Reuse => 1); + $self->{sock} = tcp_server $my_host, $my_port, sub { $self->new_client(@_); }, sub { return 256; }; die "Could not create socket: $! \n" unless $self->{sock}; - set_event_handler ($self->{sock}, read => sub { $self->new_client } ); return $self; } @@ -473,27 +305,9 @@ sub new_server { sub nolinger { my $conn = shift; - - unless ($main::is_win) { - if (isdbg('sock')) { - my ($l, $t) = unpack "ll", getsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER); - my $k = unpack 'l', getsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE); - my $n = $main::is_win ? 0 : unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY); - dbg("Linger is: $l $t, keepalive: $k, nagle: $n"); - } - - eval {setsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE, 1)} or dbg("setsockopt keepalive: $!"); - eval {setsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER, pack("ll", 0, 0))} or dbg("setsockopt linger: $!"); - eval {setsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY, 1)} or eval {setsockopt($conn->{sock}, SOL_SOCKET, TCP_NODELAY, 1)} or dbg("setsockopt tcp_nodelay: $!"); - $conn->{sock}->autoflush(0); - - if (isdbg('sock')) { - my ($l, $t) = unpack "ll", getsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER); - my $k = unpack 'l', getsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE); - my $n = $main::is_win ? 0 : unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY); - dbg("Linger is: $l $t, keepalive: $k, nagle: $n"); - } - } + my $sock = $conn->{sock}; +# $sock->linger(0); +# $sock->keepalive(1); } sub dequeue @@ -521,76 +335,70 @@ sub _rcv { # Complement to _send return unless defined($sock); my @lines; -# if ($conn->{blocking}) { -# blocking($sock, 0); -# $conn->{blocking} = 0; -# } - $bytes_read = sysread ($sock, $msg, 1024, 0); - if (defined ($bytes_read)) { - if ($bytes_read > 0) { - $total_in += $bytes_read; - if (isdbg('raw')) { - my $call = $conn->{call} || 'none'; - dbgdump('raw', "$call read $bytes_read: ", $msg); - } - if ($conn->{echo}) { - my @ch = split //, $msg; - my $out; - for (@ch) { - if (/[\cH\x7f]/) { - $out .= "\cH \cH"; - $conn->{msg} =~ s/.$//; - } else { - $out .= $_; - $conn->{msg} .= $_; - } - } - if (defined $out) { - set_event_handler ($sock, write => sub{$conn->_send(0)}); - push @{$conn->{outqueue}}, $out; + $msg = $sock->{rbuf}; + $bytes_read = length $msg || 0; + $sock->{rbuf} = ''; + + if ($bytes_read > 0) { + $total_in += $bytes_read; + if (isdbg('raw')) { + my $call = $conn->{call} || 'none'; + dbgdump('raw', "$call read $bytes_read: ", $msg); + } + if ($conn->{echo}) { + my @ch = split //, $msg; + my $out; + for (@ch) { + if (/[\cH\x7f]/) { + $out .= "\cH \cH"; + $conn->{msg} =~ s/.$//; + } else { + $out .= $_; + $conn->{msg} .= $_; } - } else { - $conn->{msg} .= $msg; } - } - } else { - if (_err_will_block($!)) { - return ; + if (defined $out) { + $conn->send_now($out); + } } else { - $bytes_read = 0; + $conn->{msg} .= $msg; } - } + } -FINISH: - if (defined $bytes_read && $bytes_read == 0) { - &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc}; - $conn->disconnect; - } else { - unless ($conn->{disable_read}) { - $conn->dequeue if exists $conn->{msg}; - } + unless ($conn->{disable_read}) { + $conn->dequeue if exists $conn->{msg}; } } sub new_client { my $server_conn = shift; - my $sock = $server_conn->{sock}->accept(); + my $sock = shift; + my $peerhost = shift; + my $peerport = shift; if ($sock) { my $conn = $server_conn->new($server_conn->{rproc}); - $conn->{sock} = $sock; - blocking($sock, 0); - $conn->nolinger; + $conn->{sock} = AnyEvent::Handle->new( + + fh => $sock, + + on_eof => sub {$conn->disconnect}, + + on_error => sub {$conn->disconnect}, + + keepalive => 1, + + linger => 0, + ); $conn->{blocking} = 0; - my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $sock->peerhost(), $conn->{peerport} = $sock->peerport()); + my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $peerhost, $conn->{peerport} = $peerport); + dbg("accept $conn->{cnum} from $conn->{peerhost} $conn->{peerport}") if isdbg('connll'); $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); + $conn->{sock}->on_read(sub {$conn->_rcv}); } else { # Login failed &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc}; $conn->disconnect(); @@ -603,8 +411,7 @@ sub new_client { sub close_server { my $conn = shift; - set_event_handler ($conn->{sock}, read => undef, write => undef, error => undef ); - $conn->{sock}->close; + undef $conn->{sock}; } # close all clients (this is for forking really) @@ -618,96 +425,31 @@ sub close_all_clients sub disable_read { my $conn = shift; - set_event_handler ($conn->{sock}, read => undef); - return $_[0] ? $conn->{disable_read} = $_[0] : $_[0]; -} - -# -#---------------------------------------------------- -# Event loop routines used by both client and server - -sub set_event_handler { - shift unless ref($_[0]); # shift if first arg is package name - my ($handle, %args) = @_; - my $callback; - if (exists $args{'write'}) { - $callback = $args{'write'}; - if ($callback) { - $wt_callbacks{$handle} = $callback; - $wt_handles->add($handle); - } else { - delete $wt_callbacks{$handle}; - $wt_handles->remove($handle); - } - } - if (exists $args{'read'}) { - $callback = $args{'read'}; - if ($callback) { - $rd_callbacks{$handle} = $callback; - $rd_handles->add($handle); - } else { - delete $rd_callbacks{$handle}; - $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, $wronly) = @_; # event_loop(1) to process events once - my ($conn, $r, $w, $e, $rset, $wset, $eset); - while (1) { - - # Quit the loop if no handles left to process - if ($wronly) { - last unless $wt_handles->count(); - - ($rset, $wset, $eset) = IO::Select->select(undef, $wt_handles, undef, $timeout); - - foreach $w (@$wset) { - &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w}; - } - } else { - - last unless ($rd_handles->count() || $wt_handles->count()); - - ($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}; - } - foreach $r (@$rset) { - &{$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; - } - } + return defined $_[0] ? $conn->{disable_read} = $_[0] : $_[0]; } sub sleep { my ($pkg, $interval) = @_; - my $now = time; - while (time - $now < $interval) { - $pkg->event_loop(10, 0.01); + my $cv = AnyEvent->condvar; + my $wait_a_bit = AnyEvent->timer( + after => $interval, + cb => sub {$cv->send}, + ); + $cv->recv; +} + +sub set_event_handler +{ + my $sock = shift; + my %args = @_; + my ($pkg, $fn, $line) = caller; + my $s; + foreach (my ($k,$v) = each %args) { + $s .= "$k => $v, "; } + $s =~ s/[\s,]$//; + dbg("Msg::set_event_handler called from ${pkg}::${fn} line $line doing $s"); } sub DESTROY diff --git a/perl/Version.pm b/perl/Version.pm index 67a15806..e3905020 100644 --- a/perl/Version.pm +++ b/perl/Version.pm @@ -9,9 +9,9 @@ package main; use vars qw($version $subversion $build $gitversion); -$version = '1.55'; +$version = '1.56'; $subversion = '0'; -$build = '100'; -$gitversion = '84fb1f6'; +$build = '1'; +$gitversion = 'd17f05b'; 1; diff --git a/perl/cluster.pl b/perl/cluster.pl index 2cf7173f..a40a5aa5 100755 --- a/perl/cluster.pl +++ b/perl/cluster.pl @@ -144,10 +144,9 @@ sub already_conn { my ($conn, $call, $mess) = @_; - $conn->disable_read(1); dbg("-> D $call $mess\n") if isdbg('chan'); + $conn->disable_read(1); $conn->send_now("D$call|$mess"); - sleep(2); $conn->disconnect; } @@ -372,6 +371,8 @@ sub idle_loop AGWMsg::process(); BPQMsg::process(); + Timer::handler(); + if (defined &Local::process) { eval { Local::process(); # do any localised processing @@ -579,8 +580,7 @@ $script->run($main::me) if $script; #open(DB::OUT, "|tee /tmp/aa"); -my $idle_loop = AnyEvent->idle(cb => &idle_loop); - +my $per_sec = AnyEvent->timer(after => 0, interval => 0.010, cb => sub{idle_loop()}); # main loop $decease->recv; diff --git a/perl/console.pl b/perl/console.pl index 0a6d7404..a25e1533 100755 --- a/perl/console.pl +++ b/perl/console.pl @@ -26,6 +26,9 @@ BEGIN { $is_win = ($^O =~ /^MS/ || $^O =~ /^OS-2/) ? 1 : 0; # is it Windows? } +use strict; + +use AnyEvent; use Msg; use IntMsg; use DXVars; @@ -39,21 +42,35 @@ use Text::Wrap; use Console; +use vars qw($maxkhist $maxshist $foreground $background $mycallcolor @colors ); + # # initialisation # -$call = ""; # the callsign being used -$conn = 0; # the connection object for the cluster -$lasttime = time; # lasttime something happened on the interface +my $call = ""; # the callsign being used +my $conn = 0; # the connection object for the cluster +my $lasttime = time; # lasttime something happened on the interface + +my $connsort = "local"; +my @khistory = (); +my @shistory = (); +my $khistpos = 0; +my $pos; +my $lth; +my $bot; +my $top; +my $pagel = 25; +my $cols = 80; +my $lines = 25; +my $scr; +my $spos = $pos = $lth = 0; +my $inbuf = ""; +my @time = (); -$connsort = "local"; -@khistory = (); -@shistory = (); -$khistpos = 0; -$spos = $pos = $lth = 0; -$inbuf = ""; -@time = (); +my $lastmin = 0; +my $sigint; +my $sigterm; #$SIG{WINCH} = sub {@time = gettimeofday}; @@ -67,16 +84,16 @@ sub mydbg sub do_initscr { $scr = new Curses; - if ($has_colors) { + if ($main::has_colors) { start_color(); - init_pair("0", $foreground, $background); -# init_pair(0, $background, $foreground); - init_pair(1, COLOR_RED, $background); - init_pair(2, COLOR_YELLOW, $background); - init_pair(3, COLOR_GREEN, $background); - init_pair(4, COLOR_CYAN, $background); - init_pair(5, COLOR_BLUE, $background); - init_pair(6, COLOR_MAGENTA, $background); + init_pair("0", $main::foreground, $main::background); +# init_pair(0, $main::background, $main::foreground); + init_pair(1, COLOR_RED, $main::background); + init_pair(2, COLOR_YELLOW, $main::background); + init_pair(3, COLOR_GREEN, $main::background); + init_pair(4, COLOR_CYAN, $main::background); + init_pair(5, COLOR_BLUE, $main::background); + init_pair(6, COLOR_MAGENTA, $main::background); init_pair(7, COLOR_RED, COLOR_BLUE); init_pair(8, COLOR_YELLOW, COLOR_BLUE); init_pair(9, COLOR_GREEN, COLOR_BLUE); @@ -85,7 +102,7 @@ sub do_initscr init_pair(12, COLOR_MAGENTA, COLOR_BLUE); init_pair(13, COLOR_YELLOW, COLOR_GREEN); init_pair(14, COLOR_RED, COLOR_GREEN); - eval { assume_default_colors($foreground, $background) } unless $is_win; + eval { assume_default_colors($main::foreground, $main::background) } unless $main::is_win; } $top = $scr->subwin($lines-4, $cols, 0, 0); @@ -117,7 +134,7 @@ sub do_resize nonl(); $lines = LINES; $cols = COLS; - $has_colors = has_colors(); + $main::has_colors = has_colors(); do_initscr(); show_screen(); @@ -143,7 +160,7 @@ sub sig_term # determine the colour of the line sub setattr { - if ($has_colors) { + if ($main::has_colors) { foreach my $ref (@colors) { if ($_[0] =~ m{$$ref[0]}) { $top->attrset($$ref[1]); @@ -176,7 +193,7 @@ sub show_screen setattr($line); $top->addstr($line); # $top->addstr("\n"); - $top->attrset(COLOR_PAIR(0)) if $has_colors; + $top->attrset(COLOR_PAIR(0)) if $main::has_colors; $spos = @shistory; } else { @@ -192,7 +209,7 @@ sub show_screen $p = 0 if $p < 0; $top->move(0, 0); - $top->attrset(COLOR_PAIR(0)) if $has_colors; + $top->attrset(COLOR_PAIR(0)) if $main::has_colors; $top->clrtobot(); for ($i = 0; $i < $pagel && $p < @shistory; $p++) { my $line = $shistory[$p]; @@ -201,7 +218,7 @@ sub show_screen $top->addstr("\n") if $i; setattr($line); $top->addstr($line); - $top->attrset(COLOR_PAIR(0)) if $has_colors; + $top->attrset(COLOR_PAIR(0)) if $main::has_colors; $i += $lines; } $spos = $p; @@ -215,9 +232,9 @@ sub show_screen $scr->addstr($lines-4, 0, $str); $scr->addstr($size); - $scr->attrset($mycallcolor) if $has_colors; + $scr->attrset($mycallcolor) if $main::has_colors; $scr->addstr($call); - $scr->attrset(COLOR_PAIR(0)) if $has_colors; + $scr->attrset(COLOR_PAIR(0)) if $main::has_colors; $scr->addstr($add); $scr->refresh(); # $top->refresh(); @@ -443,13 +460,39 @@ sub rec_stdin $bot->refresh(); } +sub idle_loop +{ + my $t; + $t = time; + if ($t > $lasttime) { + my ($min)= (gmtime($t))[1]; + if ($min != $lastmin) { + show_screen(); + $lastmin = $min; + } + $lasttime = $t; + } + my $ch = $bot->getch(); + if (@time && tv_interval(\@time, [gettimeofday]) >= 1) { +# mydbg("Got Resize"); +# do_resize(); + next; + } + if (defined $ch) { + if ($ch ne '-1') { + rec_stdin($ch); + } + } + $top->refresh() if $top->is_wintouched; + $bot->refresh(); +} # # deal with args # $call = uc shift @ARGV if @ARGV; -$call = uc $myalias if !$call; +$call = uc $main::myalias if !$call; my ($scall, $ssid) = split /-/, $call; $ssid = undef unless $ssid && $ssid =~ /^\d+$/; if ($ssid) { @@ -457,33 +500,35 @@ if ($ssid) { $call = "$scall-$ssid"; } -if ($call eq $mycall) { - print "You cannot connect as your cluster callsign ($mycall)\n"; +if ($call eq $main::mycall) { + print "You cannot connect as your cluster callsign ($main::mycall)\n"; exit(0); } dbginit(); -$conn = IntMsg->connect("$clusteraddr", $clusterport, \&rec_socket); +$conn = IntMsg->connect("$main::clusteraddr", $main::clusterport, \&rec_socket); if (! $conn) { - if (-r "$data/offline") { - open IN, "$data/offline" or die; + if (-r "$main::root/data/offline") { + open IN, "$main::root/data/offline" or die; while () { print $_; } close IN; } else { - print "Sorry, the cluster $mycall is currently off-line\n"; + print "Sorry, the cluster $main::mycall is currently off-line\n"; } exit(0); } -$conn->set_error(sub{cease(0)}); +# create end condvar +my $decease = AnyEvent->condvar; +$conn->set_error(sub{cease(0)}); unless ($DB::VERSION) { - $SIG{'INT'} = \&sig_term; - $SIG{'TERM'} = \&sig_term; + $sigint = AnyEvent->signal(signal=>'INT', cb=> sub{$decease->send}); + $sigterm = AnyEvent->signal(signal=>'TERM', cb=> sub{$decease->send}); } $SIG{'HUP'} = \&sig_term; @@ -501,32 +546,9 @@ $conn->send_later("I$call|set/page $maxshist"); $Text::Wrap::Columns = $cols; -my $lastmin = 0; -for (;;) { - my $t; - Msg->event_loop(1, 0.01); - $t = time; - if ($t > $lasttime) { - my ($min)= (gmtime($t))[1]; - if ($min != $lastmin) { - show_screen(); - $lastmin = $min; - } - $lasttime = $t; - } - my $ch = $bot->getch(); - if (@time && tv_interval(\@time, [gettimeofday]) >= 1) { -# mydbg("Got Resize"); -# do_resize(); - next; - } - if (defined $ch) { - if ($ch ne '-1') { - rec_stdin($ch); - } - } - $top->refresh() if $top->is_wintouched; - $bot->refresh(); -} +my $event_loop = AnyEvent->timer(after => 0, interval => 0.010, cb => sub{idle_loop()}); + +$decease->recv; +cease(0); exit(0); -- 2.43.0