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