[PATCH 5/7] lei: TSTP affects all curl and related subprocesses

[Eric Wong <e@80x24.org>]

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).
---
 MANIFEST                      |  1 +
 lib/PublicInbox/AutoReap.pm   | 34 +++++++++++++++++++++++++++++++++
 lib/PublicInbox/LEI.pm        |  7 +------
 lib/PublicInbox/LeiInput.pm   |  7 ++++---
 lib/PublicInbox/LeiMirror.pm  |  8 +++-----
 lib/PublicInbox/LeiRemote.pm  | 13 +++++--------
 lib/PublicInbox/LeiToMail.pm  |  2 +-
 lib/PublicInbox/LeiXSearch.pm | 20 +++++++++----------
 lib/PublicInbox/TestCommon.pm | 36 +++--------------------------------
 9 files changed, 61 insertions(+), 67 deletions(-)
 create mode 100644 lib/PublicInbox/AutoReap.pm

diff --git a/MANIFEST b/MANIFEST
index 122ceda0a761..b89513d5afb5 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 000000000000..23ecce772186
--- /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 9620e2642213..d0905562f616 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 6a90e7e1e756..dd40d83840c5 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 f1bc82e27205..a75c99c4987f 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 7782aa9dbfa1..54750062fd5f 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 5a220ba39735..9c748deaed16 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 668d0b6e5df3..fba168613d96 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 57f1db952e49..835779996d56 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]) }