public-inbox-nntpd

   1 #!/usr/bin/perl -w
   2 # Copyright (C) 2015 all contributors <meta@public-inbox.org>
   3 # License: AGPLv3 or later (https://www.gnu.org/licenses/agpl-3.0.txt)
   4 use strict;
   5 use warnings;
   6 my @CMD = ($0, @ARGV);
   7 require Danga::Socket;
   8 require IO::Handle;
   9 use Getopt::Long qw/:config gnu_getopt no_ignore_case auto_abbrev/;
  10 require PublicInbox::NewsGroup;
  11 my $set_user;
  12 my $nntpd = PublicInbox::NNTPD->new;
  13 my $refresh = sub { $nntpd->refresh_groups };
  14 $SIG{HUP} = $SIG{USR1} = $SIG{USR2} = $SIG{PIPE} =
  15         $SIG{TTIN} = $SIG{TTOU} = $SIG{WINCH} = 'IGNORE';
  16
  17 $refresh->();
  18 my (@cfg_listen, $stdout, $stderr, $group, $user, $pid_file, $daemonize);
  19 my $worker_processes = 0;
  20 my %opts = (
  21         'l|listen=s' => \@cfg_listen,
  22         '1|stdout=s' => \$stdout,
  23         '2|stderr=s' => \$stderr,
  24         'W|worker-processes=i' => \$worker_processes,
  25         'P|pid-file=s' => \$pid_file,
  26         'u|user=s' => \$user,
  27         'g|group=s' => \$group,
  28         'D|daemonize' => \$daemonize,
  29 );
  30 GetOptions(%opts) or die "bad command-line args\n";
  31
  32 if (defined $pid_file && $pid_file =~ /\.oldbin\z/) {
  33         die "--pid-file cannot end with '.oldbin'\n";
  34 }
  35
  36 my %pids;
  37 my %listener_names;
  38 my $reexec_pid;
  39 my @listeners = inherit();
  40
  41 # ignore daemonize when inheriting
  42 $daemonize = undef if scalar @listeners;
  43
  44 # default NNTP listener if no listeners
  45 push @cfg_listen, '0.0.0.0:119' unless (@listeners || @cfg_listen);
  46
  47 foreach my $l (@cfg_listen) {
  48         next if $listener_names{$l}; # already inherited
  49         require IO::Socket::INET6; # works for IPv4, too
  50         my %o = (
  51                 LocalAddr => $l,
  52                 ReuseAddr => 1,
  53                 Proto => 'tcp',
  54         );
  55         if (my $s = IO::Socket::INET6->new(%o)) {
  56                 $listener_names{sockname($s)} = $s;
  57                 push @listeners, $s;
  58         } else {
  59                 warn "error binding $l: $!\n";
  60         }
  61 }
  62 die 'No listeners bound' unless @listeners;
  63
  64 chdir '/' or die "chdir failed: $!\n";
  65 open(STDIN, '+<', '/dev/null') or die "redirect stdin failed: $!\n";
  66
  67 if (defined $pid_file || defined $group || defined $user || $daemonize) {
  68         require Net::Server::Daemonize;
  69
  70         Net::Server::Daemonize::check_pid_file($pid_file) if defined $pid_file;
  71         my $uid = Net::Server::Daemonize::get_uid($user) if defined $user;
  72         my $gid;
  73         if (defined $group) {
  74                 $gid = Net::Server::Daemonize::get_gid($group);
  75                 $gid = (split /\s+/, $gid)[0];
  76         } elsif (defined $uid) {
  77                 $gid = (getpwuid($uid))[3];
  78         }
  79
  80         # We change users in the worker to ensure upgradability,
  81         # The upgrade will create the ".oldbin" pid file in the
  82         # same directory as the given pid file.
  83         $uid and $set_user = sub {
  84                 Net::Server::Daemonize::set_user($uid, $gid);
  85         };
  86
  87         if ($daemonize) {
  88                 my ($pid, $err) = do_fork();
  89                 die "could not fork: $err\n" unless defined $pid;
  90                 exit if $pid;
  91
  92                 open STDOUT, '>&STDIN' or die "redirect stdout failed: $!\n";
  93                 open STDERR, '>&STDIN' or die "redirect stderr failed: $!\n";
  94                 POSIX::setsid();
  95                 ($pid, $err) = do_fork();
  96                 die "could not fork: $err\n" unless defined $pid;
  97                 exit if $pid;
  98         }
  99         if (defined $pid_file) {
 100                 my $unlink_pid = $$;
 101                 Net::Server::Daemonize::create_pid_file($pid_file);
 102                 END { unlink_pid_file_safe_ish($unlink_pid, $pid_file) };
 103         }
 104 }
 105
 106 if ($worker_processes > 0) {
 107         pipe(my ($p0, $p1)) or die "failed to create parent-pipe: $!\n";
 108         my %pwatch = ( fileno($p0) => sub { kill('TERM', $$) } );
 109         pipe(my ($r, $w)) or die "failed to create self-pipe: $!\n";
 110         IO::Handle::blocking($w, 0);
 111         my $set_workers = $worker_processes;
 112         my @caught;
 113         my $master_pid = $$;
 114         foreach my $s (qw(HUP CHLD QUIT INT TERM USR1 USR2 TTIN TTOU WINCH)) {
 115                 $SIG{$s} = sub {
 116                         return if $$ != $master_pid;
 117                         push @caught, $s;
 118                         syswrite($w, '.');
 119                 };
 120         }
 121         reopen_logs();
 122         # main loop
 123         while (1) {
 124                 while (my $s = shift @caught) {
 125                         if ($s eq 'USR1') {
 126                                 reopen_logs();
 127                                 kill_workers($s);
 128                         } elsif ($s eq 'USR2') {
 129                                 upgrade();
 130                         } elsif ($s =~ /\A(?:QUIT|TERM|INT)\z/) {
 131                                 # drops pipes and causes children to die
 132                                 exit
 133                         } elsif ($s eq 'WINCH') {
 134                                 $worker_processes = 0;
 135                         } elsif ($s eq 'HUP') {
 136                                 $worker_processes = $set_workers;
 137                                 $refresh->();
 138                                 kill_workers($s);
 139                         } elsif ($s eq 'TTIN') {
 140                                 if ($set_workers > $worker_processes) {
 141                                         ++$worker_processes;
 142                                 } else {
 143                                         $worker_processes = ++$set_workers;
 144                                 }
 145                         } elsif ($s eq 'TTOU') {
 146                                 if ($set_workers > 0) {
 147                                         $worker_processes = --$set_workers;
 148                                 }
 149                         } elsif ($s eq 'CHLD') {
 150                                 reap_children();
 151                         }
 152                 }
 153
 154                 my $n = scalar keys %pids;
 155                 if ($n > $worker_processes) {
 156                         while (my ($k, $v) = each %pids) {
 157                                 kill('TERM', $k) if $v >= $worker_processes;
 158                         }
 159                         $n = $worker_processes;
 160                 }
 161                 foreach my $i ($n..($worker_processes - 1)) {
 162                         my ($pid, $err) = do_fork();
 163                         if (!defined $pid) {
 164                                 warn "failed to fork worker[$i]: $err\n";
 165                         } elsif ($pid == 0) {
 166                                 $set_user->() if $set_user;
 167                                 close($_) for ($w, $r, $p1);
 168                                 Danga::Socket->AddOtherFds(%pwatch);
 169                                 goto worker;
 170                         } else {
 171                                 warn "PID=$pid is worker[$i]\n";
 172                                 $pids{$pid} = $i;
 173                         }
 174                 }
 175                 sysread($r, my $buf, 8);
 176         }
 177 } else {
 178 worker:
 179         # this calls epoll_create:
 180         @listeners = map { PublicInbox::Listener->new($_) } @listeners;
 181         reopen_logs();
 182         $SIG{QUIT} = $SIG{INT} = $SIG{TERM} = *worker_quit;
 183         $SIG{USR1} = *reopen_logs;
 184         $SIG{HUP} = $refresh;
 185         $_->watch_read(1) for @listeners;
 186         Danga::Socket->EventLoop;
 187 }
 188
 189 # end of main
 190
 191 sub worker_quit {
 192         # killing again terminates immediately:
 193         exit unless @listeners;
 194
 195         $_->close for @listeners;
 196         @listeners = ();
 197
 198         # give slow clients 30s to finish reading/writing whatever
 199         Danga::Socket->AddTimer(30, sub { exit });
 200
 201         # drop idle connections and try to quit gracefully
 202         Danga::Socket->SetPostLoopCallback(sub {
 203                 my ($dmap, undef) = @_;
 204                 my $n = 0;
 205
 206                 foreach my $s (values %$dmap) {
 207                         if ($s->can('busy') && $s->busy) {
 208                                 ++$n;
 209                         } else {
 210                                 $s->close;
 211                         }
 212                 }
 213                 $n; # true: loop continues, false: loop breaks
 214         });
 215 }
 216
 217 sub reopen_logs {
 218         if ($stdout) {
 219                 open STDOUT, '>>', $stdout or
 220                         warn "failed to redirect stdout to $stdout: $!\n";
 221         }
 222         if ($stderr) {
 223                 open STDERR, '>>', $stderr or
 224                         warn "failed to redirect stderr to $stderr: $!\n";
 225         }
 226 }
 227
 228 sub sockname {
 229         my ($s) = @_;
 230         my $n = getsockname($s) or return;
 231         my ($port, $addr);
 232         if (length($n) >= 28) {
 233                 require Socket6;
 234                 ($port, $addr) = Socket6::unpack_sockaddr_in6($n);
 235         } else {
 236                 ($port, $addr) = Socket::sockaddr_in($n);
 237         }
 238         if (length($addr) == 4) {
 239                 $n = Socket::inet_ntoa($addr)
 240         } else {
 241                 $n = '['.Socket6::inet_ntop(Socket6::AF_INET6(), $addr).']';
 242         }
 243         $n .= ":$port";
 244 }
 245
 246 sub inherit {
 247         return () if ($ENV{LISTEN_PID} || 0) != $$;
 248         my $fds = $ENV{LISTEN_FDS} or return ();
 249         my $end = $fds + 2; # LISTEN_FDS_START - 1
 250         my @rv = ();
 251         foreach my $fd (3..$end) {
 252                 my $s = IO::Handle->new;
 253                 $s->fdopen($fd, 'r');
 254                 if (my $k = sockname($s)) {
 255                         $listener_names{$k} = $s;
 256                         push @rv, $s;
 257                 } else {
 258                         warn "failed to inherit fd=$fd (LISTEN_FDS=$fds)";
 259                 }
 260         }
 261         @rv
 262 }
 263
 264 sub upgrade {
 265         if ($reexec_pid) {
 266                 warn "upgrade in-progress: $reexec_pid\n";
 267                 return;
 268         }
 269         if (defined $pid_file) {
 270                 if ($pid_file =~ /\.oldbin\z/) {
 271                         warn "BUG: .oldbin suffix exists: $pid_file\n";
 272                         return;
 273                 }
 274                 unlink_pid_file_safe_ish($$, $pid_file);
 275                 $pid_file .= '.oldbin';
 276                 Net::Server::Daemonize::create_pid_file($pid_file);
 277         }
 278         my ($pid, $err) = do_fork();
 279         unless (defined $pid) {
 280                 warn "fork failed: $err\n";
 281                 return;
 282         }
 283         if ($pid == 0) {
 284                 use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD);
 285                 $ENV{LISTEN_FDS} = scalar @listeners;
 286                 $ENV{LISTEN_PID} = $$;
 287                 foreach my $s (@listeners) {
 288                         my $fl = fcntl($s, F_GETFD, 0);
 289                         fcntl($s, F_SETFD, $fl &= ~FD_CLOEXEC);
 290                 }
 291                 exec @CMD;
 292                 die "Failed to exec: $!\n";
 293         }
 294         $reexec_pid = $pid;
 295 }
 296
 297 sub kill_workers {
 298         my ($s) = @_;
 299
 300         while (my ($pid, $id) = each %pids) {
 301                 kill $s, $pid;
 302         }
 303 }
 304
 305 sub do_fork {
 306         require POSIX;
 307         my $new = POSIX::SigSet->new;
 308         $new->fillset;
 309         my $old = POSIX::SigSet->new;
 310         POSIX::sigprocmask(&POSIX::SIG_BLOCK, $new, $old) or
 311                                 die "SIG_BLOCK: $!\n";
 312         my $pid = fork;
 313         my $err = $!;
 314         POSIX::sigprocmask(&POSIX::SIG_SETMASK, $old) or
 315                                 die "SIG_SETMASK: $!\n";
 316         ($pid, $err);
 317 }
 318
 319 sub upgrade_aborted ($) {
 320         my ($p) = @_;
 321         warn "reexec PID($p) died with: $?\n";
 322         $reexec_pid = undef;
 323         return unless $pid_file;
 324
 325         my $file = $pid_file;
 326         $file =~ s/\.oldbin\z// or die "BUG: no '.oldbin' suffix in $file\n";
 327         unlink_pid_file_safe_ish($$, $pid_file);
 328         $pid_file = $file;
 329         eval { Net::Server::Daemonize::create_pid_file($pid_file) };
 330         warn $@, "\n" if $@;
 331 }
 332
 333 sub reap_children {
 334         while (1) {
 335                 my $p = waitpid(-1, &POSIX::WNOHANG) or return;
 336                 if (defined $reexec_pid && $p == $reexec_pid) {
 337                         upgrade_aborted($p);
 338                 } elsif (defined(my $id = delete $pids{$p})) {
 339                         warn "worker[$id] PID($p) died with: $?\n";
 340                 } elsif ($p > 0) {
 341                         warn "unknown PID($p) reaped: $?\n";
 342                 } else {
 343                         return;
 344                 }
 345         }
 346 }
 347
 348 sub unlink_pid_file_safe_ish ($$) {
 349         my ($unlink_pid, $file) = @_;
 350         return unless defined $unlink_pid && $unlink_pid == $$;
 351
 352         open my $fh, '<', $file or return;
 353         defined(my $read_pid = <$fh>) or return;
 354         chomp $read_pid;
 355         if ($read_pid == $unlink_pid) {
 356                 Net::Server::Daemonize::unlink_pid_file($file);
 357         }
 358 }
 359
 360 1;
 361 package PublicInbox::Listener;
 362 use strict;
 363 use warnings;
 364 use base 'Danga::Socket';
 365 use Socket qw(SOL_SOCKET SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
 366 use PublicInbox::NNTP;
 367
 368 sub new ($$) {
 369         my ($class, $s) = @_;
 370         setsockopt($s, SOL_SOCKET, SO_KEEPALIVE, 1);
 371         setsockopt($s, IPPROTO_TCP, TCP_NODELAY, 1);
 372         listen($s, 1024);
 373         IO::Handle::blocking($s, 0);
 374         my $self = fields::new($class);
 375         $self->SUPER::new($s);
 376 }
 377
 378 sub event_read {
 379         my ($self) = @_;
 380         # no loop here, we want to fairly distribute clients
 381         # between multiple processes sharing the same socket
 382         if (accept(my $c, $self->{sock})) {
 383                 IO::Handle::blocking($c, 0); # no accept4 :<
 384                 PublicInbox::NNTP->new($c, $nntpd);
 385         }
 386 }
 387
 388 1;
 389 package PublicInbox::NNTPD;
 390 use strict;
 391 use warnings;
 392 use fields qw(groups grouplist err out);
 393
 394 sub new {
 395         my ($class) = @_;
 396         my $self = fields::new($class);
 397         $self->{groups} = {};
 398         $self->{err} = \*STDERR;
 399         $self->{out} = \*STDOUT;
 400         $self->{grouplist} = [];
 401         $self;
 402 }
 403
 404 sub refresh_groups {
 405         my ($self) = @_;
 406         require PublicInbox::Config;
 407         my $pi_config = PublicInbox::Config->new;
 408         my $new = {};
 409         my @list;
 410         foreach my $k (keys %$pi_config) {
 411                 $k =~ /\Apublicinbox\.([^\.]+)\.mainrepo\z/ or next;
 412                 my $g = $1;
 413                 my $git_dir = $pi_config->{$k};
 414                 my $addr = $pi_config->{"publicinbox.$g.address"};
 415                 my $ngname = $pi_config->{"publicinbox.$g.newsgroup"};
 416                 if (defined $ngname) {
 417                         next if ($ngname eq ''); # disabled
 418                         $g = $ngname;
 419                 }
 420                 my $ng = PublicInbox::NewsGroup->new($g, $git_dir, $addr);
 421                 my $old_ng = $self->{groups}->{$g};
 422
 423                 # Reuse the old one if possible since it can hold
 424                 # references to valid mm and gcf objects
 425                 if ($old_ng) {
 426                         $old_ng->update($ng);
 427                         $ng = $old_ng;
 428                 }
 429
 430                 # Only valid if Msgmap works
 431                 if ($ng->mm(1)) {
 432                         $new->{$g} = $ng;
 433                         push @list, $ng;
 434                 }
 435         }
 436         @list = sort { $a->{name} cmp $b->{name} } @list;
 437         $self->{grouplist} = \@list;
 438         # this will destroy old groups that got deleted
 439         %{$self->{groups}} = %$new;
 440 }
 441
 442 1;