[svn:qpsmtpd] rev 413 - in branches/high_perf: . lib/Danga lib/Danga/DNS lib/Qpsmtpd plugins
[email protected] 9 May 2005 13:43:40 -0000
| Newsgroups | perl.cvs.qpsmtpd |
|---|---|
| Message-ID | <[email protected]> |
Author: msergeant
Date: Mon May 9 06:43:40 2005
New Revision: 413
Modified:
branches/high_perf/lib/Danga/Client.pm
branches/high_perf/lib/Danga/DNS/Resolver.pm
branches/high_perf/lib/Danga/Socket.pm
branches/high_perf/lib/Danga/TimeoutSocket.pm
branches/high_perf/lib/Qpsmtpd/ConfigServer.pm
branches/high_perf/lib/Qpsmtpd/Plugin.pm
branches/high_perf/lib/Qpsmtpd/PollServer.pm
branches/high_perf/plugins/check_earlytalker
branches/high_perf/qpsmtpd
Log:
Large number of patches from Brian Grossman to fix a number of bugs
Implement connection timeout
Modified: branches/high_perf/lib/Danga/Client.pm
==============================================================================
--- branches/high_perf/lib/Danga/Client.pm (original)
+++ branches/high_perf/lib/Danga/Client.pm Mon May 9 06:43:40 2005
@@ -6,7 +6,8 @@
use Time::HiRes ();
# 30 seconds max timeout!
-sub max_idle_time { 30 }
+sub max_idle_time { 30 }
+sub max_connect_time { 1200 }
sub new {
my Danga::Client $self = shift;
@@ -45,7 +46,7 @@
my Danga::Client $self = shift;
my ($timeout) = @_;
my $end = Time::HiRes::time() + $timeout;
- warn("Calling can-read\n");
+ # warn("Calling can-read\n");
$self->{can_read_mode} = 1;
if (!length($self->{line})) {
my $old = $self->watch_read();
@@ -61,7 +62,7 @@
$self->SetPostLoopCallback(sub { $self->have_line ? 0 : 1 });
return if $self->{closing};
$self->{alive_time} = time;
- warn("can_read returning for '$self->{line}'\n");
+ # warn("can_read returning for '$self->{line}'\n");
return 1 if length($self->{line});
return;
}
Modified: branches/high_perf/lib/Danga/DNS/Resolver.pm
==============================================================================
--- branches/high_perf/lib/Danga/DNS/Resolver.pm (original)
+++ branches/high_perf/lib/Danga/DNS/Resolver.pm Mon May 9 06:43:40 2005
@@ -286,6 +286,9 @@
#$self->{timeout}{$id} = time();
}
+ elsif ($err eq "NOERROR") {
+ $asker->run_callback($err, $query);
+ }
elsif($err) {
print("error: $err\n");
$asker->run_callback($err, $query);
Modified: branches/high_perf/lib/Danga/Socket.pm
==============================================================================
--- branches/high_perf/lib/Danga/Socket.pm (original)
+++ branches/high_perf/lib/Danga/Socket.pm Mon May 9 06:43:40 2005
@@ -24,7 +24,7 @@
$VERSION = do { my @r = (q$Revision: 1.4 $ =~ /\d+/g); sprintf "%d."."%02d" x $#r, @r };
use fields qw(sock fd write_buf write_buf_offset write_buf_size
- read_push_back
+ read_push_back post_loop_callback
closed event_watch debug_level);
use Errno qw(EINPROGRESS EWOULDBLOCK EISCONN
@@ -307,9 +307,21 @@
# now we can close sockets that wanted to close during our event processing.
# (we didn't want to close them during the loop, as we didn't want fd numbers
# being reused and confused during the event loop)
- $_->close while ($_ = shift @ToClose);
+ while(my $j = shift @ToClose) {
+ $j->[1]->close();
+ $j->[0]->{closing} = 0;
+ }
+
+
+ # now we're at the very end, call per-connection callbacks if defined
+ for my $fd (%DescriptorMap) {
+ my $pob = $DescriptorMap{$fd};
+ if( defined $pob->{post_loop_callback} ) {
+ return unless $pob->{post_loop_callback}->(\%DescriptorMap, \%OtherFds);
+ }
+ }
- # now we're at the very end, call callback if defined
+ # now we're at the very end, call global callback if defined
if (defined $PostLoopCallback) {
return $PostLoopCallback->(\%DescriptorMap, \%OtherFds);
}
@@ -401,6 +413,7 @@
$self->{write_buf_size} = 0;
$self->{closed} = 0;
$self->{read_push_back} = [];
+ $self->{post_loop_callback} = undef;
$self->{event_watch} = POLLERR|POLLHUP|POLLNVAL;
@@ -472,7 +485,7 @@
# defer closing the actual socket until the event loop is done
# processing this round of events. (otherwise we might reuse fds)
- push @ToClose, $sock;
+ push @ToClose, [$self,$sock];
return 0;
}
@@ -785,7 +798,18 @@
### be passed two parameters: \%DescriptorMap, \%OtherFds.
sub SetPostLoopCallback {
my ($class, $ref) = @_;
- $PostLoopCallback = (defined $ref && ref $ref eq 'CODE') ? $ref : undef;
+ if(ref $class) {
+ my Danga::Socket $self = $class;
+ if( defined $ref && ref $ref eq 'CODE' ) {
+ $self->{PostLoopCallback} = $ref;
+ }
+ else {
+ delete $self->{PostLoopCallback};
+ }
+ }
+ else {
+ $PostLoopCallback = (defined $ref && ref $ref eq 'CODE') ? $ref : undef;
+ }
}
#####################################################################
Modified: branches/high_perf/lib/Danga/TimeoutSocket.pm
==============================================================================
--- branches/high_perf/lib/Danga/TimeoutSocket.pm (original)
+++ branches/high_perf/lib/Danga/TimeoutSocket.pm Mon May 9 06:43:40 2005
@@ -24,22 +24,46 @@
return $self;
}
+sub ticker {
+ my Danga::TimeoutSocket $self = shift;
+
+ my $now = time;
+
+ if ($now - 15 > $last_cleanup) {
+ $last_cleanup = $now;
+ _do_cleanup($now);
+ }
+}
+
+# overload these in a subclass
+sub max_idle_time { 0 }
+sub max_connect_time { 0 }
+
sub _do_cleanup {
my $now = shift;
my $sf = __PACKAGE__->get_sock_ref;
my %max_age; # classname -> max age (0 means forever)
+ my %max_connect; # classname -> max connect time
my @to_close;
while (my $k = each %$sf) {
my Danga::TimeoutSocket $v = $sf->{$k};
my $ref = ref $v;
next unless $v->isa('Danga::TimeoutSocket');
unless (defined $max_age{$ref}) {
- $max_age{$ref} = $ref->max_idle_time || 0;
+ $max_age{$ref} = $ref->max_idle_time || 0;
+ $max_connect{$ref} = $ref->max_connect_time || 0;
+ }
+ if (my $t = $max_connect{$ref}) {
+ if ($v->{create_time} < $now - $t) {
+ push @to_close, $v;
+ next;
+ }
}
- next unless $max_age{$ref};
- if ($v->{alive_time} < $now - $max_age{$ref}) {
- push @to_close, $v;
+ if (my $t = $max_age{$ref}) {
+ if ($v->{alive_time} < $now - $t) {
+ push @to_close, $v;
+ }
}
}
Modified: branches/high_perf/lib/Qpsmtpd/ConfigServer.pm
==============================================================================
--- branches/high_perf/lib/Qpsmtpd/ConfigServer.pm (original)
+++ branches/high_perf/lib/Qpsmtpd/ConfigServer.pm Mon May 9 06:43:40 2005
@@ -15,6 +15,7 @@
_transaction
_test_mode
_extras
+ other_fds
);
my $PROMPT = "Enter command: ";
@@ -130,6 +131,16 @@
return "PAUSED";
}
+sub cmd_continue {
+ my $self = shift;
+
+ my $other_fds = $self->{other_fds};
+
+ $self->OtherFds( %$other_fds );
+ %$other_fds = ();
+ return "UNPAUSED";
+}
+
sub cmd_status {
my $self = shift;
@@ -173,7 +184,7 @@
}
}
- $output .= "Curr Connections: $current_connections\n".
+ $output .= "Curr Connections: $current_connections / $::MAXconn\n".
"Curr DNS Queries: $current_dns";
return $output;
@@ -206,7 +217,7 @@
}
}
foreach my $item (@all) {
- $list .= sprintf("%x : %s [%s] Connected %0.2fs\n", @$item);
+ $list .= sprintf("%x : %s [%s] Connected %0.2fs\n", map { defined()?$_:'' } @$item);
}
return $list;
Modified: branches/high_perf/lib/Qpsmtpd/Plugin.pm
==============================================================================
--- branches/high_perf/lib/Qpsmtpd/Plugin.pm (original)
+++ branches/high_perf/lib/Qpsmtpd/Plugin.pm Mon May 9 06:43:40 2005
@@ -108,7 +108,7 @@
}
close F;
- my $line = "\n#line 1 $file\n";
+ my $line = "\n#line 0 $file\n";
if ($test_mode) {
if (open(F, "t/plugin_tests/$plugin")) {
Modified: branches/high_perf/lib/Qpsmtpd/PollServer.pm
==============================================================================
--- branches/high_perf/lib/Qpsmtpd/PollServer.pm (original)
+++ branches/high_perf/lib/Qpsmtpd/PollServer.pm Mon May 9 06:43:40 2005
@@ -32,6 +32,9 @@
use Time::HiRes qw(time);
use strict;
+sub max_idle_time { 60 }
+sub max_connect_time { 1200 }
+
sub input_sock {
my $self = shift;
@_ and $self->{input_sock} = shift;
@@ -91,7 +94,7 @@
if ($::DEBUG > 1) { print "$$:".($self+0)."C($self->{mode}): $line"; }
local $SIG{ALRM} = sub {
my ($pkg, $file, $line) = caller();
- die "ALARM: $pkg, $file, $line";
+ die "ALARM: ($self->{mode}) $pkg, $file, $line";
};
my $prev = alarm(2); # must process a command in < 2 seconds
eval { $self->_process_line($line) };
@@ -169,6 +172,7 @@
my ($ip, $port) = split(':', $self->peer_addr_string);
$conn->remote_ip($ip);
$conn->remote_port($port);
+ $conn->remote_info("[$ip]");
Danga::DNS->new(
client => $self,
# NB: Setting remote_info to the same as remote_host
Modified: branches/high_perf/plugins/check_earlytalker
==============================================================================
--- branches/high_perf/plugins/check_earlytalker (original)
+++ branches/high_perf/plugins/check_earlytalker Mon May 9 06:43:40 2005
@@ -44,8 +44,6 @@
=cut
-use Time::HiRes ();
-
use warnings;
use strict;
@@ -70,25 +68,19 @@
sub connect_handler {
my ($self, $transaction) = @_;
- my $qp = $self->qp;
- my $end = Time::HiRes::time + $self->{_args}->{'wait'} ;
- my $time;
- for( $time = Time::HiRes::time; $time < $end && !length($qp->{line}) ; $time = Time::HiRes::time ) {
- $qp->can_read($end-$time);
- }
- my $earlytalker = 0;
- $earlytalker = 1 if $time < $end ;
-
- if ($earlytalker) {
+
+ if ($self->qp->can_read($self->{_args}->{'wait'})) {
$self->log(LOGNOTICE, 'remote host started talking before we said hello');
if ($self->{_args}->{'defer-reject'}) {
- $self->connection->notes('earlytalker', 1);
- } else {
+ $self->connection->notes('earlytalker', 1);
+ }
+ else {
my $msg = 'Connecting host started transmitting before SMTP greeting';
return (DENY,$msg) if $self->{_args}->{'action'} eq 'deny';
return (DENYSOFT,$msg) if $self->{_args}->{'action'} eq 'denysoft';
}
- } else {
+ }
+ else {
$self->log(LOGINFO, 'remote host said nothing spontaneous, proceeding');
}
return DECLINED;
Modified: branches/high_perf/qpsmtpd
==============================================================================
--- branches/high_perf/qpsmtpd (original)
+++ branches/high_perf/qpsmtpd Mon May 9 06:43:40 2005
@@ -199,6 +199,7 @@
}
sub run_as_server {
+ local $::MAXconn = $MAXCONN;
# establish SERVER socket, bind and listen.
$SERVER = IO::Socket::INET->new(LocalPort => $PORT,
LocalAddr => $LOCALADDR,
@@ -290,11 +291,19 @@
# Accept a new connection
sub accept_handler {
- my $running = scalar keys %childstatus;
+ my $running;
+ if( $LineMode ) {
+ $running = scalar keys %childstatus;
+ }
+ else {
+ my $descriptors = Danga::Client->DescriptorMap;
+ $running = scalar keys %$descriptors;
+ }
while ($running >= $MAXCONN) {
::log(LOGINFO,"Too many connections: $running >= $MAXCONN.");
return;
}
+ ++$running if $LineMode; # count self
my $csock = $SERVER->accept();
if (!$csock) {
@@ -341,6 +350,7 @@
$client->close;
return;
}
+ ::log(LOGINFO, "accepted connection $running/$MAXCONN ($num_conn/$MAXCONNIP) from $rem_ip");
}
my $rc = $client->start_conversation;