lei: TSTP affects all curl and related subprocesses

By relying more on pgroups for remaining remaining processes,
this lets us pause all curl+tail subprocesses with a single
kill(2) to avoid cluttering stderr.

We won't bother pausing the pigz/gzip/bzip2/xz compressor
process not cat-file processes, though, since those don't write
to the terminal and they idle soon after the workers react to
SIGSTOP.

AutoReap is hoisted out from TestCommon.pm.  CLONE_SKIP
is gone since we won't be using Perl threads any time
soon (they're discouraged by the maintainers of Perl).

9 files changed, 61 insertions, 67 deletions

diff --git a/MANIFEST b/MANIFEST
index 122ceda0..b89513d5 100644
--- a/MANIFEST
+++ b/MANIFEST
@@ -147,6 +147,7 @@ lib/PublicInbox/AddressPP.pm
 lib/PublicInbox/Admin.pm
 lib/PublicInbox/AdminEdit.pm
 lib/PublicInbox/AltId.pm
+lib/PublicInbox/AutoReap.pm
 lib/PublicInbox/Cgit.pm
 lib/PublicInbox/CmdIPC4.pm
 lib/PublicInbox/CompressNoop.pm
diff --git a/lib/PublicInbox/AutoReap.pm b/lib/PublicInbox/AutoReap.pm
new file mode 100644
index 00000000..23ecce77
--- /dev/null
+++ b/lib/PublicInbox/AutoReap.pm
@@ -0,0 +1,34 @@
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# automatically kill + reap children when this goes out-of-scope
+package PublicInbox::AutoReap;
+use v5.10.1;
+use strict;
+
+sub new {
+        my (undef, $pid, $cb) = @_;
+        bless { pid => $pid, cb => $cb, owner => $$ }, __PACKAGE__
+}
+
+sub kill {
+        my ($self, $sig) = @_;
+        CORE::kill($sig // 'TERM', $self->{pid});
+}
+
+sub join {
+        my ($self, $sig) = @_;
+        my $pid = delete $self->{pid} or return;
+        $self->{cb}->() if defined $self->{cb};
+        CORE::kill($sig, $pid) if defined $sig;
+        my $ret = waitpid($pid, 0) // die "waitpid($pid): $!";
+        $ret == $pid or die "BUG: waitpid($pid) != $ret";
+}
+
+sub DESTROY {
+        my ($self) = @_;
+        return if $self->{owner} != $$;
+        $self->join('TERM');
+}
+
+1;
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 9620e264..d0905562 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -516,12 +516,6 @@ sub sigpipe_handler { # handles SIGPIPE from @WQ_KEYS workers
         fail_handler($_[0], 13, delete $_[0]->{1});
 }
 
-# PublicInbox::OnDestroy callback for SIGINT to take out the entire pgid
-sub sigint_reap {
-        my ($pgid) = @_;
-        dwaitpid($pgid) if kill('-INT', $pgid);
-}
-
 sub fail ($$;$) {
         my ($self, $buf, $exit_code) = @_;
         local $current_lei = $self;
@@ -600,6 +594,7 @@ sub _lei_atfork_child {
                         $cb->(@_) unless PublicInbox::Eml::warn_ignore(@_)
                 };
         }
+        $SIG{TERM} = sub { exit(128 + 15) };
         $current_lei = $persist ? undef : $self; # for SIG{__WARN__}
 }
 
diff --git a/lib/PublicInbox/LeiInput.pm b/lib/PublicInbox/LeiInput.pm
index 6a90e7e1..dd40d838 100644
--- a/lib/PublicInbox/LeiInput.pm
+++ b/lib/PublicInbox/LeiInput.pm
@@ -8,6 +8,7 @@ use v5.10.1;
 use PublicInbox::DS;
 use PublicInbox::Spawn qw(which popen_rd);
 use PublicInbox::InboxWritable qw(eml_from_path);
+use PublicInbox::AutoReap;
 
 # JMAP RFC 8621 4.1.1
 # https://www.iana.org/assignments/imap-jmap-keywords/imap-jmap-keywords.xhtml
@@ -102,13 +103,13 @@ sub handle_http_input ($$@) {
         push @$curl, '-s', @$curl_opt;
         my $cmd = $curl->for_uri($lei, $uri);
         $lei->qerr("# $cmd");
-        my $rdr = { 2 => $lei->{2}, pgid => 0 };
-        my ($fh, $pid) = popen_rd($cmd, undef, $rdr);
+        my ($fh, $pid) = popen_rd($cmd, undef, { 2 => $lei->{2} });
+        my $ar = PublicInbox::AutoReap->new($pid);
         grep(/\A--compressed\z/, @$curl) or
                 $fh = IO::Uncompress::Gunzip->new($fh, MultiStream => 1);
         eval { $self->input_fh('mboxrd', $fh, $url, @args) };
         my $err = $@;
-        waitpid($pid, 0);
+        $ar->join;
         $? || $err and
                 $lei->child_error($?, "@$cmd failed".$err ? " $err" : '');
 }
diff --git a/lib/PublicInbox/LeiMirror.pm b/lib/PublicInbox/LeiMirror.pm
index f1bc82e2..a75c99c4 100644
--- a/lib/PublicInbox/LeiMirror.pm
+++ b/lib/PublicInbox/LeiMirror.pm
@@ -7,6 +7,7 @@ use strict;
 use v5.10.1;
 use parent qw(PublicInbox::IPC);
 use PublicInbox::Config;
+use PublicInbox::AutoReap;
 use IO::Uncompress::Gunzip qw(gunzip $GunzipError);
 use IO::Compress::Gzip qw(gzip $GzipError);
 use PublicInbox::Spawn qw(popen_rd spawn);
@@ -192,10 +193,8 @@ sub index_cloned_inbox {
 sub run_reap {
         my ($lei, $cmd, $opt) = @_;
         $lei->qerr("# @$cmd");
-        my $pid = spawn($cmd, undef, $opt);
-        my $reap = PublicInbox::OnDestroy->new($lei->can('sigint_reap'), $pid);
-        waitpid($pid, 0) == $pid or die "waitpid @$cmd: $!";
-        @$reap = (); # cancel reap
+        my $ar = PublicInbox::AutoReap->new(spawn($cmd, undef, $opt));
+        $ar->join;
         my $ret = $?;
         $? = 0; # don't let it influence normal exit
         $ret;
@@ -459,7 +458,6 @@ sub start {
 sub ipc_atfork_child {
         my ($self) = @_;
         $self->{lei}->_lei_atfork_child;
-        $SIG{TERM} = sub { exit(128 + 15) }; # trigger OnDestroy $reap
         $self->SUPER::ipc_atfork_child;
 }
 
diff --git a/lib/PublicInbox/LeiRemote.pm b/lib/PublicInbox/LeiRemote.pm
index 7782aa9d..54750062 100644
--- a/lib/PublicInbox/LeiRemote.pm
+++ b/lib/PublicInbox/LeiRemote.pm
@@ -9,10 +9,10 @@ package PublicInbox::LeiRemote;
 use v5.10.1;
 use strict;
 use IO::Uncompress::Gunzip;
-use PublicInbox::OnDestroy;
 use PublicInbox::MboxReader;
 use PublicInbox::Spawn qw(popen_rd);
 use PublicInbox::LeiCurl;
+use PublicInbox::AutoReap;
 use PublicInbox::ContentHash qw(git_sha);
 
 sub new {
@@ -47,17 +47,14 @@ sub mset {
         $uri->query_form(q => $qstr, x => 'm', r => 1); # r=1: relevance
         my $cmd = $curl->for_uri($self->{lei}, $uri);
         $self->{lei}->qerr("# $cmd");
-        my $rdr = { 2 => $lei->{2}, pgid => 0 };
-        my ($fh, $pid) = popen_rd($cmd, undef, $rdr);
-        my $reap = PublicInbox::OnDestroy->new($lei->can('sigint_reap'), $pid);
+        my ($fh, $pid) = popen_rd($cmd, undef, { 2 => $lei->{2} });
+        my $ar = PublicInbox::AutoReap->new($pid);
         $self->{smsg} = [];
         $fh = IO::Uncompress::Gunzip->new($fh, MultiStream => 1);
         PublicInbox::MboxReader->mboxrd($fh, \&_each_mboxrd_eml, $self);
-        my $err = waitpid($pid, 0) == $pid ? undef
-                                        : "BUG: waitpid($cmd): $!";
-        @$reap = (); # cancel OnDestroy
         my $wait = $self->{lei}->{sto}->wq_do('done');
-        die $err if $err;
+        $ar->join;
+        $lei->child_error($?) if $?;
         $self; # we are the mset (and $ibx, and $self)
 }
 
diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index 5a220ba3..9c748dea 100644
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -157,7 +157,7 @@ sub _post_augment_mbox { # open a compressor process from top-level process
         my $zsfx = $self->{zsfx} or return;
         my $cmd = PublicInbox::MboxReader::zsfx2cmd($zsfx, undef, $lei);
         my ($r, $w) = @{delete $lei->{zpipe}};
-        my $rdr = { 0 => $r, 1 => $lei->{1}, 2 => $lei->{2} };
+        my $rdr = { 0 => $r, 1 => $lei->{1}, 2 => $lei->{2}, pgid => 0 };
         my $pid = spawn($cmd, undef, $rdr);
         my $pp = gensym;
         my $dup = bless { "pid.$pid" => $cmd }, ref($lei);
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 668d0b6e..fba16861 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -15,6 +15,7 @@ use PublicInbox::Search qw(xap_terms);
 use PublicInbox::Spawn qw(popen_rd spawn which);
 use PublicInbox::MID qw(mids);
 use PublicInbox::Smsg;
+use PublicInbox::AutoReap;
 use PublicInbox::Eml;
 use PublicInbox::LEI;
 use Fcntl qw(SEEK_SET F_SETFL O_APPEND O_RDWR);
@@ -346,18 +347,17 @@ sub query_remote_mboxrd {
         my @qform = (x => 'm');
         push(@qform, t => 1) if $opt->{threads};
         my $verbose = $opt->{verbose};
-        my ($reap_tail, $reap_curl);
+        my $reap_tail;
         my $cerr = File::Temp->new(TEMPLATE => 'curl.err-XXXX', TMPDIR => 1);
         fcntl($cerr, F_SETFL, O_APPEND|O_RDWR) or warn "set O_APPEND: $!";
-        my $rdr = { 2 => $cerr, pgid => 0 };
-        my $sigint_reap = $lei->can('sigint_reap');
+        my $rdr = { 2 => $cerr };
         if ($verbose) {
                 # spawn a process to force line-buffering, otherwise curl
                 # will write 1 character at-a-time and parallel outputs
                 # mmmaaayyy llloookkk llliiikkkeee ttthhhiiisss
-                my $o = { 1 => $lei->{2}, 2 => $lei->{2}, pgid => 0 };
+                my $o = { 1 => $lei->{2}, 2 => $lei->{2} };
                 my $pid = spawn(['tail', '-f', $cerr->filename], undef, $o);
-                $reap_tail = PublicInbox::OnDestroy->new($sigint_reap, $pid);
+                $reap_tail = PublicInbox::AutoReap->new($pid);
         }
         my $curl = PublicInbox::LeiCurl->new($lei, $self->{curl}) or return;
         push @$curl, '-s', '-d', '';
@@ -372,16 +372,13 @@ sub query_remote_mboxrd {
                 my $cmd = $curl->for_uri($lei, $uri);
                 $lei->qerr("# $cmd");
                 my ($fh, $pid) = popen_rd($cmd, undef, $rdr);
-                $reap_curl = PublicInbox::OnDestroy->new($sigint_reap, $pid);
+                my $reap_curl = PublicInbox::AutoReap->new($pid);
                 $fh = IO::Uncompress::Gunzip->new($fh, MultiStream => 1);
                 PublicInbox::MboxReader->mboxrd($fh, \&each_remote_eml, $self,
                                                 $lei, $each_smsg);
-                my $err = waitpid($pid, 0) == $pid ? undef
-                                                : "BUG: waitpid($cmd): $!";
-                @$reap_curl = (); # cancel OnDestroy
-                die $err if $err;
                 my $nr = $lei->{-nr_remote_eml};
                 my $wait = $lei->{sto}->wq_do('done') if $nr && $lei->{sto};
+                $reap_curl->join;
                 if ($? == 0) {
                         # don't update if no results, maybe MTA is down
                         $key && $nr and
@@ -389,7 +386,7 @@ sub query_remote_mboxrd {
                         mset_progress($lei, $lei->{-current_url}, $nr, $nr);
                         next;
                 }
-                $err = '';
+                my $err;
                 if (-s $cerr) {
                         seek($cerr, 0, SEEK_SET) //
                                         warn "seek($cmd stderr): $!";
@@ -397,6 +394,7 @@ sub query_remote_mboxrd {
                                         warn "read($cmd stderr): $!";
                         truncate($cerr, 0) // warn "truncate($cmd stderr): $!";
                 }
+                $err //= '';
                 next if (($? >> 8) == 22 && $err =~ /\b404\b/);
                 $uri->query_form(q => $qstr);
                 $lei->child_error($?, "E: <$uri> $err");
diff --git a/lib/PublicInbox/TestCommon.pm b/lib/PublicInbox/TestCommon.pm
index 57f1db95..83577999 100644
--- a/lib/PublicInbox/TestCommon.pm
+++ b/lib/PublicInbox/TestCommon.pm
@@ -6,6 +6,7 @@ package PublicInbox::TestCommon;
 use strict;
 use parent qw(Exporter);
 use v5.10.1;
+use PublicInbox::AutoReap;
 use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD :seek);
 use POSIX qw(dup2);
 use IO::Socket::INET;
@@ -429,7 +430,7 @@ sub tail_f (@) {
         require PublicInbox::Spawn;
         my $pid = PublicInbox::Spawn::spawn($cmd, undef, { 1 => 2 });
         wait_for_tail($pid, scalar @_);
-        PublicInboxTestProcess->new($pid, \&wait_for_tail);
+        PublicInbox::AutoReap->new($pid, \&wait_for_tail);
 }
 
 sub start_script {
@@ -492,7 +493,7 @@ sub start_script {
                         die "FAIL: ",join(' ', $key, @argv), ": $!\n";
                 }
         }
-        my $td = PublicInboxTestProcess->new($pid);
+        my $td = PublicInbox::AutoReap->new($pid);
         $td->{-extra} = $tail;
         $td;
 }
@@ -742,37 +743,6 @@ sub test_httpd ($$;$) {
 };
 
 
-package PublicInboxTestProcess;
-use strict;
-
-# prevent new threads from inheriting these objects
-sub CLONE_SKIP { 1 }
-
-sub new {
-        my ($cls, $pid, $cb) = @_;
-        bless { pid => $pid, cb => $cb, owner => $$ }, $cls;
-}
-
-sub kill {
-        my ($self, $sig) = @_;
-        CORE::kill($sig // 'TERM', $self->{pid});
-}
-
-sub join {
-        my ($self, $sig) = @_;
-        my $pid = delete $self->{pid} or return;
-        $self->{cb}->() if defined $self->{cb};
-        CORE::kill($sig, $pid) if defined $sig;
-        my $ret = waitpid($pid, 0) // die "waitpid($pid): $!";
-        $ret == $pid or die "waitpid($pid) != $ret";
-}
-
-sub DESTROY {
-        my ($self) = @_;
-        return if $self->{owner} != $$;
-        $self->join('TERM');
-}
-
 package PublicInbox::TestCommon::InboxWakeup;
 use strict;
 sub on_inbox_unlock { ${$_[0]}->($_[1]) }