byterazor/qpsmtpd
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Don't listen for readiness in the parent any more - breaks under high load.

Browse Source 
git-svn-id: https://svn.perl.org/qpsmtpd/trunk@823 958fd67b-6ff1-0310-b445-bb7760255be9
This commit is contained in:
Matt Sergeant 2007-12-03 21:37:45 +00:00
parent 40c485e769
commit cfa23dedec
1 changed files with 19 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
qpsmtpd-async
View File

@ -100,10 +100,11 @@ my $POLL = "with " . ($Danga::Socket::HaveEpoll ? "epoll()" :
my $SERVER; my $SERVER;
my $CONFIG_SERVER; my $CONFIG_SERVER;
use constant ACCEPTING => 1; use constant READY => 1;
use constant RESTARTING => 2; use constant ACCEPTING => 2;
use constant RESTARTING => 999;
my %childstatus = (); my %childstatus = ();
my %childhandle = ();
if ($PID_FILE && -r $PID_FILE) { if ($PID_FILE && -r $PID_FILE) {
open PID, "<$PID_FILE" open PID, "<$PID_FILE"
@ -150,21 +151,22 @@ sub spawn_child {
my $plugin_loader = shift || Qpsmtpd::SMTP->new; my $plugin_loader = shift || Qpsmtpd::SMTP->new;
socketpair(my $reader, my $writer, AF_UNIX, SOCK_STREAM, PF_UNSPEC) || die "Unable to create a pipe"; socketpair(my $reader, my $writer, AF_UNIX, SOCK_STREAM, PF_UNSPEC) || die "Unable to create a pipe";
$reader->autoflush(1);
$writer->autoflush(1); $writer->autoflush(1);
$reader->autoflush(1);
if (my $pid = _fork) { if (my $pid = _fork) {
$childstatus{$pid} = ACCEPTING; $childstatus{$pid} = $writer;
$childhandle{$pid} = $writer;
return $pid; return $pid;
} }
$SIG{CHLD} = $SIG{INT} = $SIG{TERM} = 'DEFAULT'; $SIG{CHLD} = $SIG{INT} = $SIG{TERM} = 'DEFAULT';
$SIG{PIPE} = 'IGNORE'; $SIG{PIPE} = 'IGNORE';
$SIG{HUP} = sub { print "Child got SIGHUP\n" }; $SIG{HUP} = 'IGNORE';
# sub { cmd_hup(); Qpsmtpd::PollServer->EventLoop(); }; # so we can HUP just one child
Qpsmtpd::PollServer->OtherFds(fileno($reader) => sub { command_handler($reader) }); Qpsmtpd::PollServer->OtherFds(
fileno($reader) => sub { command_handler($reader) },
fileno($SERVER) => \&accept_handler,
);
$plugin_loader->run_hooks('post-fork'); $plugin_loader->run_hooks('post-fork');
@ -172,13 +174,11 @@ sub spawn_child {
exit; exit;
} }
# Note this is broken on KQueue because it requires that it handle signals itself or it breaks the event loop.
sub sig_hup { sub sig_hup {
for my $writer (values %childhandle) { for my $writer (values %childstatus) {
print $writer "hup\n"; print $writer "hup\n";
my $result = <$writer>;
} }
$SIG{HUP} = \&sig_hup;
Qpsmtpd::PollServer->EventLoop();
} }
sub sig_chld { sub sig_chld {
@ -191,7 +191,6 @@ sub sig_chld {
last unless $child > 0; last unless $child > 0;
print "SIGCHLD: child $child died\n"; print "SIGCHLD: child $child died\n";
delete $childstatus{$child}; delete $childstatus{$child};
delete $childhandle{$child};
$spawn_count++; $spawn_count++;
} }
if ($spawn_count) { if ($spawn_count) {
@ -282,9 +281,9 @@ sub run_as_server {
$SIG{HUP} = \&sig_hup; $SIG{HUP} = \&sig_hup;
Qpsmtpd::PollServer->OtherFds( Qpsmtpd::PollServer->OtherFds(
fileno($SERVER) => \&accept_handler, fileno($CONFIG_SERVER) => \&config_handler,
fileno($CONFIG_SERVER) => \&config_handler );
);
Qpsmtpd::PollServer->EventLoop; Qpsmtpd::PollServer->EventLoop;
exit; exit;
@ -309,20 +308,6 @@ sub config_handler {
return; return;
} }
# server is ready to accept - tell a child to accept().
sub accept_handler {
# pick a random child to tell to accept()
my $child = (shuffle keys %childstatus)[0];
if ($childstatus{$child} != ACCEPTING) {
# recurse...
return accept_handler() if %childstatus;
die "No children available";
}
my $writer = $childhandle{$child};
print $writer "accept\n";
my $result = <$writer>;
}
sub command_handler { sub command_handler {
my $reader = shift; my $reader = shift;
@ -333,20 +318,18 @@ sub command_handler {
my $real_command = "cmd_$command"; my $real_command = "cmd_$command";
no strict 'refs'; no strict 'refs';
my $result = $real_command->(); $real_command->();
print $reader "$result\n";
} }
sub cmd_hup { sub cmd_hup {
# clear cache # clear cache
#print "Clearing cache\n"; print "Clearing cache\n";
Qpsmtpd::clear_config_cache(); Qpsmtpd::clear_config_cache();
# should also reload modules... but can't do that yet. # should also reload modules... but can't do that yet.
return "ok";
} }
# Accept all new connections # Accept all new connections
sub cmd_accept { sub accept_handler {
for (1 .. $NUMACCEPT) { for (1 .. $NUMACCEPT) {
return unless _accept_handler(); return unless _accept_handler();
} }
@ -357,7 +340,6 @@ sub cmd_accept {
$NUMACCEPT = ACCEPT_MAX if $NUMACCEPT > ACCEPT_MAX; $NUMACCEPT = ACCEPT_MAX if $NUMACCEPT > ACCEPT_MAX;
$ACCEPT_RSET->cancel; $ACCEPT_RSET->cancel;
$ACCEPT_RSET = Danga::Socket->AddTimer(30, \&reset_num_accept); $ACCEPT_RSET = Danga::Socket->AddTimer(30, \&reset_num_accept);
return "ok";
} }
use Errno qw(EAGAIN EWOULDBLOCK); use Errno qw(EAGAIN EWOULDBLOCK);