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                 if ($uid and !chown($uid, $gid, $pid_file)) {
 103                         warn "could not chown $pid_file: $!\n";
 104                 }
 105                 END { unlink_pid_file_safe_ish($unlink_pid, $pid_file) };
 106         }
 107 }
 108
 109 if ($worker_processes > 0) {
 110         pipe(my ($p0, $p1)) or die "failed to create parent-pipe: $!\n";
 111         my %pwatch = ( fileno($p0) => sub { kill('TERM', $$) } );
 112         pipe(my ($r, $w)) or die "failed to create self-pipe: $!\n";
 113         IO::Handle::blocking($w, 0);
 114         my $set_workers = $worker_processes;
 115         my @caught;
 116         my $master_pid = $$;
 117         foreach my $s (qw(HUP CHLD QUIT INT TERM USR1 USR2 TTIN TTOU WINCH)) {
 118                 $SIG{$s} = sub {
 119                         return if $$ != $master_pid;
 120                         push @caught, $s;
 121                         syswrite($w, '.');
 122                 };
 123         }
 124         reopen_logs();
 125         # main loop
 126         while (1) {
 127                 while (my $s = shift @caught) {
 128                         if ($s eq 'USR1') {
 129                                 reopen_logs();
 130                                 kill_workers($s);
 131                         } elsif ($s eq 'USR2') {
 132                                 upgrade();
 133                         } elsif ($s =~ /\A(?:QUIT|TERM|INT)\z/) {
 134                                 # drops pipes and causes children to die
 135                                 exit
 136                         } elsif ($s eq 'WINCH') {
 137                                 $worker_processes = 0;
 138                         } elsif ($s eq 'HUP') {
 139                                 $worker_processes = $set_workers;
 140                                 $refresh->();
 141                                 kill_workers($s);
 142                         } elsif ($s eq 'TTIN') {
 143                                 if ($set_workers > $worker_processes) {
 144                                         ++$worker_processes;
 145                                 } else {
 146                                         $worker_processes = ++$set_workers;
 147                                 }
 148                         } elsif ($s eq 'TTOU') {
 149                                 if ($set_workers > 0) {
 150                                         $worker_processes = --$set_workers;
 151                                 }
 152                         } elsif ($s eq 'CHLD') {
 153                                 reap_children();
 154                         }
 155                 }
 156
 157                 my $n = scalar keys %pids;
 158                 if ($n > $worker_processes) {
 159                         while (my ($k, $v) = each %pids) {
 160                                 kill('TERM', $k) if $v >= $worker_processes;
 161                         }
 162                         $n = $worker_processes;
 163                 }
 164                 foreach my $i ($n..($worker_processes - 1)) {
 165                         my ($pid, $err) = do_fork();
 166                         if (!defined $pid) {
 167                                 warn "failed to fork worker[$i]: $err\n";
 168                         } elsif ($pid == 0) {
 169                                 $set_user->() if $set_user;
 170                                 close($_) for ($w, $r, $p1);
 171                                 Danga::Socket->AddOtherFds(%pwatch);
 172                                 goto worker;
 173                         } else {
 174                                 warn "PID=$pid is worker[$i]\n";
 175                                 $pids{$pid} = $i;
 176                         }
 177                 }
 178                 sysread($r, my $buf, 8);
 179         }
 180 } else {
 181         $set_user->() if $set_user;
 182         $SIG{USR2} = sub { worker_quit() if upgrade() };
 183 worker:
 184         # this calls epoll_create:
 185         reopen_logs();
 186         $SIG{QUIT} = $SIG{INT} = $SIG{TERM} = *worker_quit;
 187         $SIG{USR1} = *reopen_logs;
 188         $SIG{HUP} = $refresh;
 189         PublicInbox::Listener->new($_) for @listeners;
 190         Danga::Socket->EventLoop;
 191 }
 192
 193 # end of main
 194
 195 sub worker_quit {
 196         # killing again terminates immediately:
 197         exit unless @listeners;
 198
 199         @listeners = ();
 200
 201         # give slow clients 30s to finish reading/writing whatever
 202         Danga::Socket->AddTimer(30, sub { exit });
 203
 204         # drop idle connections and try to quit gracefully
 205         Danga::Socket->SetPostLoopCallback(sub {
 206                 my ($dmap, undef) = @_;
 207                 my $n = 0;
 208
 209                 foreach my $s (values %$dmap) {
 210                         if ($s->can('busy') && $s->busy) {
 211                                 ++$n;
 212                         } else {
 213                                 $s->close;
 214                         }
 215                 }
 216                 $n; # true: loop continues, false: loop breaks
 217         });
 218 }
 219
 220 sub reopen_logs {
 221         if ($stdout) {
 222                 open STDOUT, '>>', $stdout or
 223                         warn "failed to redirect stdout to $stdout: $!\n";
 224         }
 225         if ($stderr) {
 226                 open STDERR, '>>', $stderr or
 227                         warn "failed to redirect stderr to $stderr: $!\n";
 228         }
 229 }
 230
 231 sub sockname {
 232         my ($s) = @_;
 233         my $n = getsockname($s) or return;
 234         my ($port, $addr);
 235         if (length($n) >= 28) {
 236                 require Socket6;
 237                 ($port, $addr) = Socket6::unpack_sockaddr_in6($n);
 238         } else {
 239                 ($port, $addr) = Socket::sockaddr_in($n);
 240         }
 241         if (length($addr) == 4) {
 242                 $n = Socket::inet_ntoa($addr)
 243         } else {
 244                 $n = '['.Socket6::inet_ntop(Socket6::AF_INET6(), $addr).']';
 245         }
 246         $n .= ":$port";
 247 }
 248
 249 sub inherit {
 250         return () if ($ENV{LISTEN_PID} || 0) != $$;
 251         my $fds = $ENV{LISTEN_FDS} or return ();
 252         my $end = $fds + 2; # LISTEN_FDS_START - 1
 253         my @rv = ();
 254         foreach my $fd (3..$end) {
 255                 my $s = IO::Handle->new;
 256                 $s->fdopen($fd, 'r');
 257                 if (my $k = sockname($s)) {
 258                         $listener_names{$k} = $s;
 259                         push @rv, $s;
 260                 } else {
 261                         warn "failed to inherit fd=$fd (LISTEN_FDS=$fds)";
 262                 }
 263         }
 264         @rv
 265 }
 266
 267 sub upgrade {
 268         if ($reexec_pid) {
 269                 warn "upgrade in-progress: $reexec_pid\n";
 270                 return;
 271         }
 272         if (defined $pid_file) {
 273                 if ($pid_file =~ /\.oldbin\z/) {
 274                         warn "BUG: .oldbin suffix exists: $pid_file\n";
 275                         return;
 276                 }
 277                 unlink_pid_file_safe_ish($$, $pid_file);
 278                 $pid_file .= '.oldbin';
 279                 Net::Server::Daemonize::create_pid_file($pid_file);
 280         }
 281         my ($pid, $err) = do_fork();
 282         unless (defined $pid) {
 283                 warn "fork failed: $err\n";
 284                 return;
 285         }
 286         if ($pid == 0) {
 287                 use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD);
 288                 $ENV{LISTEN_FDS} = scalar @listeners;
 289                 $ENV{LISTEN_PID} = $$;
 290                 foreach my $s (@listeners) {
 291                         my $fl = fcntl($s, F_GETFD, 0);
 292                         fcntl($s, F_SETFD, $fl &= ~FD_CLOEXEC);
 293                 }
 294                 exec @CMD;
 295                 die "Failed to exec: $!\n";
 296         }
 297         $reexec_pid = $pid;
 298 }
 299
 300 sub kill_workers {
 301         my ($s) = @_;
 302
 303         while (my ($pid, $id) = each %pids) {
 304                 kill $s, $pid;
 305         }
 306 }
 307
 308 sub do_fork {
 309         require POSIX;
 310         my $new = POSIX::SigSet->new;
 311         $new->fillset;
 312         my $old = POSIX::SigSet->new;
 313         POSIX::sigprocmask(&POSIX::SIG_BLOCK, $new, $old) or
 314                                 die "SIG_BLOCK: $!\n";
 315         my $pid = fork;
 316         my $err = $!;
 317         POSIX::sigprocmask(&POSIX::SIG_SETMASK, $old) or
 318                                 die "SIG_SETMASK: $!\n";
 319         ($pid, $err);
 320 }
 321
 322 sub upgrade_aborted ($) {
 323         my ($p) = @_;
 324         warn "reexec PID($p) died with: $?\n";
 325         $reexec_pid = undef;
 326         return unless $pid_file;
 327
 328         my $file = $pid_file;
 329         $file =~ s/\.oldbin\z// or die "BUG: no '.oldbin' suffix in $file\n";
 330         unlink_pid_file_safe_ish($$, $pid_file);
 331         $pid_file = $file;
 332         eval { Net::Server::Daemonize::create_pid_file($pid_file) };
 333         warn $@, "\n" if $@;
 334 }
 335
 336 sub reap_children {
 337         while (1) {
 338                 my $p = waitpid(-1, &POSIX::WNOHANG) or return;
 339                 if (defined $reexec_pid && $p == $reexec_pid) {
 340                         upgrade_aborted($p);
 341                 } elsif (defined(my $id = delete $pids{$p})) {
 342                         warn "worker[$id] PID($p) died with: $?\n";
 343                 } elsif ($p > 0) {
 344                         warn "unknown PID($p) reaped: $?\n";
 345                 } else {
 346                         return;
 347                 }
 348         }
 349 }
 350
 351 sub unlink_pid_file_safe_ish ($$) {
 352         my ($unlink_pid, $file) = @_;
 353         return unless defined $unlink_pid && $unlink_pid == $$;
 354
 355         open my $fh, '<', $file or return;
 356         defined(my $read_pid = <$fh>) or return;
 357         chomp $read_pid;
 358         if ($read_pid == $unlink_pid) {
 359                 Net::Server::Daemonize::unlink_pid_file($file);
 360         }
 361 }
 362
 363 1;
 364 package PublicInbox::Listener;
 365 use strict;
 366 use warnings;
 367 use base 'Danga::Socket';
 368 use Socket qw(SOL_SOCKET SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
 369 use PublicInbox::NNTP;
 370
 371 sub new ($$) {
 372         my ($class, $s) = @_;
 373         setsockopt($s, SOL_SOCKET, SO_KEEPALIVE, 1);
 374         setsockopt($s, IPPROTO_TCP, TCP_NODELAY, 1);
 375         listen($s, 1024);
 376         IO::Handle::blocking($s, 0);
 377         my $self = fields::new($class);
 378         $self->SUPER::new($s); # calls epoll_create for the first socket
 379         $self->watch_read(1);
 380         $self
 381 }
 382
 383 sub event_read {
 384         my ($self) = @_;
 385         # no loop here, we want to fairly distribute clients
 386         # between multiple processes sharing the same socket
 387         if (accept(my $c, $self->{sock})) {
 388                 IO::Handle::blocking($c, 0); # no accept4 :<
 389                 PublicInbox::NNTP->new($c, $nntpd);
 390         }
 391 }
 392
 393 1;
 394 package PublicInbox::NNTPD;
 395 use strict;
 396 use warnings;
 397 use fields qw(groups grouplist err out);
 398
 399 sub new {
 400         my ($class) = @_;
 401         my $self = fields::new($class);
 402         $self->{groups} = {};
 403         $self->{err} = \*STDERR;
 404         $self->{out} = \*STDOUT;
 405         $self->{grouplist} = [];
 406         $self;
 407 }
 408
 409 sub refresh_groups {
 410         my ($self) = @_;
 411         require PublicInbox::Config;
 412         my $pi_config = PublicInbox::Config->new;
 413         my $new = {};
 414         my @list;
 415         foreach my $k (keys %$pi_config) {
 416                 $k =~ /\Apublicinbox\.([^\.]+)\.mainrepo\z/ or next;
 417                 my $g = $1;
 418                 my $git_dir = $pi_config->{$k};
 419                 my $addr = $pi_config->{"publicinbox.$g.address"};
 420                 my $ngname = $pi_config->{"publicinbox.$g.newsgroup"};
 421                 if (defined $ngname) {
 422                         next if ($ngname eq ''); # disabled
 423                         $g = $ngname;
 424                 }
 425                 my $ng = PublicInbox::NewsGroup->new($g, $git_dir, $addr);
 426                 my $old_ng = $self->{groups}->{$g};
 427
 428                 # Reuse the old one if possible since it can hold
 429                 # references to valid mm and gcf objects
 430                 if ($old_ng) {
 431                         $old_ng->update($ng);
 432                         $ng = $old_ng;
 433                 }
 434
 435                 # Only valid if Msgmap works
 436                 if ($ng->mm(1)) {
 437                         $new->{$g} = $ng;
 438                         push @list, $ng;
 439                 }
 440         }
 441         @list = sort { $a->{name} cmp $b->{name} } @list;
 442         $self->{grouplist} = \@list;
 443         # this will destroy old groups that got deleted
 444         %{$self->{groups}} = %$new;
 445 }
 446
 447 1;