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                         next unless ref($s) eq 'PublicInbox::NNTP';
 208                         if ($s->{write_buf_size} || $s->{rbuf}) {
 209                                 ++$n;
 210                         } else {
 211                                 $s->close;
 212                         }
 213                 }
 214                 $n; # true: loop continues, false: loop breaks
 215         });
 216 }
 217
 218 sub reopen_logs {
 219         if ($stdout) {
 220                 open STDOUT, '>>', $stdout or
 221                         warn "failed to redirect stdout to $stdout: $!\n";
 222         }
 223         if ($stderr) {
 224                 open STDERR, '>>', $stderr or
 225                         warn "failed to redirect stderr to $stderr: $!\n";
 226         }
 227 }
 228
 229 sub sockname {
 230         my ($s) = @_;
 231         my $n = getsockname($s) or return;
 232         my ($port, $addr);
 233         if (length($n) >= 28) {
 234                 require Socket6;
 235                 ($port, $addr) = Socket6::unpack_sockaddr_in6($n);
 236         } else {
 237                 ($port, $addr) = Socket::sockaddr_in($n);
 238         }
 239         if (length($addr) == 4) {
 240                 $n = Socket::inet_ntoa($addr)
 241         } else {
 242                 $n = '['.Socket6::inet_ntop(Socket6::AF_INET6(), $addr).']';
 243         }
 244         $n .= ":$port";
 245 }
 246
 247 sub inherit {
 248         return () if ($ENV{LISTEN_PID} || 0) != $$;
 249         my $fds = $ENV{LISTEN_FDS} or return ();
 250         my $end = $fds + 2; # LISTEN_FDS_START - 1
 251         my @rv = ();
 252         foreach my $fd (3..$end) {
 253                 my $s = IO::Handle->new;
 254                 $s->fdopen($fd, 'r');
 255                 if (my $k = sockname($s)) {
 256                         $listener_names{$k} = $s;
 257                         push @rv, $s;
 258                 } else {
 259                         warn "failed to inherit fd=$fd (LISTEN_FDS=$fds)";
 260                 }
 261         }
 262         @rv
 263 }
 264
 265 sub upgrade {
 266         if ($reexec_pid) {
 267                 warn "upgrade in-progress: $reexec_pid\n";
 268                 return;
 269         }
 270         if (defined $pid_file) {
 271                 if ($pid_file =~ /\.oldbin\z/) {
 272                         warn "BUG: .oldbin suffix exists: $pid_file\n";
 273                         return;
 274                 }
 275                 unlink_pid_file_safe_ish($$, $pid_file);
 276                 $pid_file .= '.oldbin';
 277                 Net::Server::Daemonize::create_pid_file($pid_file);
 278         }
 279         my ($pid, $err) = do_fork();
 280         unless (defined $pid) {
 281                 warn "fork failed: $err\n";
 282                 return;
 283         }
 284         if ($pid == 0) {
 285                 use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD);
 286                 $ENV{LISTEN_FDS} = scalar @listeners;
 287                 $ENV{LISTEN_PID} = $$;
 288                 foreach my $s (@listeners) {
 289                         my $fl = fcntl($s, F_GETFD, 0);
 290                         fcntl($s, F_SETFD, $fl &= ~FD_CLOEXEC);
 291                 }
 292                 exec @CMD;
 293                 die "Failed to exec: $!\n";
 294         }
 295         $reexec_pid = $pid;
 296 }
 297
 298 sub kill_workers {
 299         my ($s) = @_;
 300
 301         while (my ($pid, $id) = each %pids) {
 302                 kill $s, $pid;
 303         }
 304 }
 305
 306 sub do_fork {
 307         require POSIX;
 308         my $new = POSIX::SigSet->new;
 309         $new->fillset;
 310         my $old = POSIX::SigSet->new;
 311         POSIX::sigprocmask(&POSIX::SIG_BLOCK, $new, $old) or
 312                                 die "SIG_BLOCK: $!\n";
 313         my $pid = fork;
 314         my $err = $!;
 315         POSIX::sigprocmask(&POSIX::SIG_SETMASK, $old) or
 316                                 die "SIG_SETMASK: $!\n";
 317         ($pid, $err);
 318 }
 319
 320 sub upgrade_aborted ($) {
 321         my ($p) = @_;
 322         warn "reexec PID($p) died with: $?\n";
 323         $reexec_pid = undef;
 324         return unless $pid_file;
 325
 326         my $file = $pid_file;
 327         $file =~ s/\.oldbin\z// or die "BUG: no '.oldbin' suffix in $file\n";
 328         unlink_pid_file_safe_ish($$, $pid_file);
 329         $pid_file = $file;
 330         eval { Net::Server::Daemonize::create_pid_file($pid_file) };
 331         warn $@, "\n" if $@;
 332 }
 333
 334 sub reap_children {
 335         while (1) {
 336                 my $p = waitpid(-1, &POSIX::WNOHANG) or return;
 337                 if (defined $reexec_pid && $p == $reexec_pid) {
 338                         upgrade_aborted($p);
 339                 } elsif (defined(my $id = delete $pids{$p})) {
 340                         warn "worker[$id] PID($p) died with: $?\n";
 341                 } elsif ($p > 0) {
 342                         warn "unknown PID($p) reaped: $?\n";
 343                 } else {
 344                         return;
 345                 }
 346         }
 347 }
 348
 349 sub unlink_pid_file_safe_ish ($$) {
 350         my ($unlink_pid, $file) = @_;
 351         return unless defined $unlink_pid && $unlink_pid == $$;
 352
 353         open my $fh, '<', $file or return;
 354         defined(my $read_pid = <$fh>) or return;
 355         chomp $read_pid;
 356         if ($read_pid == $unlink_pid) {
 357                 Net::Server::Daemonize::unlink_pid_file($file);
 358         }
 359 }
 360
 361 1;
 362 package PublicInbox::Listener;
 363 use strict;
 364 use warnings;
 365 use base 'Danga::Socket';
 366 use Socket qw(SOL_SOCKET SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
 367 use PublicInbox::NNTP;
 368
 369 sub new ($$) {
 370         my ($class, $s) = @_;
 371         setsockopt($s, SOL_SOCKET, SO_KEEPALIVE, 1);
 372         setsockopt($s, IPPROTO_TCP, TCP_NODELAY, 1);
 373         listen($s, 1024);
 374         IO::Handle::blocking($s, 0);
 375         my $self = fields::new($class);
 376         $self->SUPER::new($s);
 377 }
 378
 379 sub event_read {
 380         my ($self) = @_;
 381         # no loop here, we want to fairly distribute clients
 382         # between multiple processes sharing the same socket
 383         if (accept(my $c, $self->{sock})) {
 384                 IO::Handle::blocking($c, 0); # no accept4 :<
 385                 PublicInbox::NNTP->new($c, $nntpd);
 386         }
 387 }
 388
 389 1;
 390 package PublicInbox::NNTPD;
 391 use strict;
 392 use warnings;
 393 use fields qw(groups grouplist err out);
 394
 395 sub new {
 396         my ($class) = @_;
 397         my $self = fields::new($class);
 398         $self->{groups} = {};
 399         $self->{err} = \*STDERR;
 400         $self->{out} = \*STDOUT;
 401         $self->{grouplist} = [];
 402         $self;
 403 }
 404
 405 sub refresh_groups {
 406         my ($self) = @_;
 407         require PublicInbox::Config;
 408         my $pi_config = PublicInbox::Config->new;
 409         my $new = {};
 410         my @list;
 411         foreach my $k (keys %$pi_config) {
 412                 $k =~ /\Apublicinbox\.([^\.]+)\.mainrepo\z/ or next;
 413                 my $g = $1;
 414                 my $git_dir = $pi_config->{$k};
 415                 my $addr = $pi_config->{"publicinbox.$g.address"};
 416                 my $ngname = $pi_config->{"publicinbox.$g.newsgroup"};
 417                 if (defined $ngname) {
 418                         next if ($ngname eq ''); # disabled
 419                         $g = $ngname;
 420                 }
 421                 my $ng = PublicInbox::NewsGroup->new($g, $git_dir, $addr);
 422                 my $old_ng = $self->{groups}->{$g};
 423
 424                 # Reuse the old one if possible since it can hold
 425                 # references to valid mm and gcf objects
 426                 if ($old_ng) {
 427                         $old_ng->update($ng);
 428                         $ng = $old_ng;
 429                 }
 430
 431                 # Only valid if Msgmap works
 432                 if ($ng->mm(1)) {
 433                         $new->{$g} = $ng;
 434                         push @list, $ng;
 435                 }
 436         }
 437         @list = sort { $a->{name} cmp $b->{name} } @list;
 438         $self->{grouplist} = \@list;
 439         # this will destroy old groups that got deleted
 440         %{$self->{groups}} = %$new;
 441 }
 442
 443 1;