[PATCH 1/3] process_pipe2 + import

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

---
 MANIFEST                        |  1 +
 lib/PublicInbox/Import.pm       | 21 +++++++--------------
 lib/PublicInbox/ProcessPipe.pm  | 15 +++++++++------
 lib/PublicInbox/ProcessPipe2.pm | 29 +++++++++++++++++++++++++++++
 4 files changed, 46 insertions(+), 20 deletions(-)
 create mode 100644 lib/PublicInbox/ProcessPipe2.pm

diff --git a/MANIFEST b/MANIFEST
index 4693cbe0..2b180f72 100644
--- a/MANIFEST
+++ b/MANIFEST
@@ -319,6 +319,7 @@ lib/PublicInbox/POP3.pm
 lib/PublicInbox/POP3D.pm
 lib/PublicInbox/PktOp.pm
 lib/PublicInbox/ProcessPipe.pm
+lib/PublicInbox/ProcessPipe2.pm
 lib/PublicInbox/Qspawn.pm
 lib/PublicInbox/Reply.pm
 lib/PublicInbox/RepoAtom.pm
diff --git a/lib/PublicInbox/Import.pm b/lib/PublicInbox/Import.pm
index 59462e9a..c63b369f 100644
--- a/lib/PublicInbox/Import.pm
+++ b/lib/PublicInbox/Import.pm
@@ -7,7 +7,7 @@
 # requires read-only access.
 package PublicInbox::Import;
 use strict;
-use parent qw(PublicInbox::Lock);
+use parent qw(PublicInbox::Lock PublicInbox::ProcessPipe2);
 use v5.10.1;
 use PublicInbox::Spawn qw(run_die popen_rd);
 use PublicInbox::MID qw(mids mid2path);
@@ -58,9 +58,6 @@ sub gfi_start {
 
 	return ($self->{in}, $self->{out}) if $self->{in};
 
-	my ($in_r, $out_r, $out_w);
-	pipe($out_r, $out_w) or die "pipe failed: $!";
-
 	$self->lock_acquire;
 	eval {
 		my ($git, $ref) = @$self{qw(git ref)};
@@ -72,18 +69,15 @@ sub gfi_start {
 			die "fatal: ls-tree -r -z --name-only $ref: \$?=$?" if $?;
 			$self->{-tree} = { map { $_ => 1 } split(/\0/, $t) };
 		}
-		$in_r = $self->{in} = $git->popen(qw(fast-import
-					--quiet --done --date-format=raw),
-					undef, { 0 => $out_r });
-		$out_w->autoflush(1);
-		$self->{out} = $out_w;
+		$self->popen_rw(['git', "--git-dir=$git->{git_dir}",
+			qw(fast-import --quiet --done --date-format=raw)]);
 		$self->{nchg} = 0;
 	};
 	if ($@) {
 		$self->lock_release;
 		die $@;
 	}
-	($in_r, $out_w);
+	@$self{qw(in out)};
 }
 
 sub wfail () { die "write to fast-import failed: $!" }
@@ -490,11 +484,10 @@ sub active { !!$_[0]->{out} }
 
 sub done {
 	my ($self) = @_;
-	my $w = delete $self->{out} or return;
+	$self->{out} or return;
 	eval {
-		my $r = delete $self->{in} or die 'BUG: missing {in} when done';
-		print $w "done\n" or wfail;
-		close $r or die "fast-import failed: $?"; # ProcessPipe::CLOSE
+		print { $self->{out} } "done\n" or wfail;
+		$self->close_wait or die "fast-import failed: $?";
 	};
 	my $wait_err = $@;
 	my $nchg = delete $self->{nchg};
diff --git a/lib/PublicInbox/ProcessPipe.pm b/lib/PublicInbox/ProcessPipe.pm
index 16971801..c82ba0f8 100644
--- a/lib/PublicInbox/ProcessPipe.pm
+++ b/lib/PublicInbox/ProcessPipe.pm
@@ -1,10 +1,9 @@
 # Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
-# a tied handle for auto reaping of children tied to a read-only pipe, see perltie(1)
-# DO NOT use this as-is for bidirectional pipes/sockets (e.g. in PublicInbox::Git),
-# both ends of the pipe must be at the same level of the Perl object hierarchy
-# to ensure orderly destruction.
+# a tied handle for auto reaping of children tied to a read-only pipe,
+# see perltie(1).  Use ProcessPipe2 for bidirectional pipes/sockets
+# for proper refcount and destruction ordering but no tie support
 package PublicInbox::ProcessPipe;
 use v5.12;
 use PublicInbox::DS qw(awaitpid);
@@ -57,8 +56,12 @@ sub FILENO { fileno($_[0]->{fh}) }
 
 sub _close ($;$) {
 	my ($self, $wait) = @_;
-	my ($fh, $pid) = delete(@$self{qw(fh pid)});
-	my $ret = defined($fh) ? close($fh) : '';
+	my ($fh, $pid, $in, $out) = delete(@$self{qw(fh pid in out)});
+	my $ret = defined($fh) ? close($fh) : do { # in/out => ProcessPipe2
+		my $n = (defined($out) ? close($out) : 1).
+			(defined($in) ? close($in) : 1);
+		$n eq '11' ? 1 : '';
+	};
 	return $ret unless defined($pid) && $self->{ppid} == $$;
 	if ($wait) { # caller cares about the exit status:
 		# synchronous wait via defined(wantarray) on awaitpid:
diff --git a/lib/PublicInbox/ProcessPipe2.pm b/lib/PublicInbox/ProcessPipe2.pm
new file mode 100644
index 00000000..d89a2b65
--- /dev/null
+++ b/lib/PublicInbox/ProcessPipe2.pm
@@ -0,0 +1,29 @@
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# non-tied version of ProcessPipe for "bidirectional" pipes used by
+# `cat-file --batch-*', fast-import, etc..
+package PublicInbox::ProcessPipe2;
+use v5.12;
+use parent qw(PublicInbox::ProcessPipe);
+use autodie qw(pipe);
+use PublicInbox::Spawn qw(spawn);
+use PublicInbox::DS qw(awaitpid);
+
+sub popen_rw {
+	my ($self, $cmd, $env, $opt) = @_;
+	pipe($self->{in}, local $opt->{1});
+	pipe(local $opt->{0}, $self->{out});
+	$self->{out}->autoflush(1);
+	$self->{ppid} = $$;
+	$self->{pp_chld_err} = \(my $err);
+	my $pid = $self->{pid} = spawn($cmd, $env, $opt);
+	awaitpid($pid, $self->can('waitcb'), \$err, @{$opt->{cb_arg} // []});
+	undef;
+}
+
+sub close_wait { $_[0]->_close(1) }
+
+# DESTROY is inherited from ProcessPipe
+
+1;