v2writable: cleanup unused pipes in partitions

Leaking these pipes to child processes wasn't harmful, but
made determining relationships and dataflow between processes
more confusing.

4 files changed, 31 insertions, 9 deletions

diff --git a/lib/PublicInbox/Import.pm b/lib/PublicInbox/Import.pm
index b650e4ef..ac46c0cb 100644
--- a/lib/PublicInbox/Import.pm
+++ b/lib/PublicInbox/Import.pm
@@ -372,6 +372,13 @@ sub done {
         close $lockfh or die "close lock failed: $!";
 }
 
+sub atfork_child {
+        my ($self) = @_;
+        foreach my $f (qw(in out)) {
+                close $self->{$f} or die "failed to close import[$f]: $!\n";
+        }
+}
+
 1;
 __END__
 =pod
diff --git a/lib/PublicInbox/SearchIdxPart.pm b/lib/PublicInbox/SearchIdxPart.pm
index 5582d672..64e52636 100644
--- a/lib/PublicInbox/SearchIdxPart.pm
+++ b/lib/PublicInbox/SearchIdxPart.pm
@@ -14,10 +14,7 @@ sub new {
         my $pid = fork;
         defined $pid or die "fork failed: $!\n";
         if ($pid == 0) {
-                foreach my $other (@{$v2writable->{idx_parts}}) {
-                        my $other_w = $other->{w} or next;
-                        close $other_w or die "close other failed: $!\n";
-                }
+                $v2writable->atfork_child;
                 $v2writable = undef;
                 close $w;
 
@@ -74,4 +71,8 @@ sub index_raw {
         $w->flush or die "failed to flush: $!\n";
 }
 
+sub atfork_child {
+        close $_[0]->{w} or die "failed to close write pipe: $!\n";
+}
+
 1;
diff --git a/lib/PublicInbox/SearchIdxThread.pm b/lib/PublicInbox/SearchIdxThread.pm
index 6471309e..7df07a65 100644
--- a/lib/PublicInbox/SearchIdxThread.pm
+++ b/lib/PublicInbox/SearchIdxThread.pm
@@ -7,8 +7,8 @@ use base qw(PublicInbox::SearchIdx);
 use Storable qw(freeze thaw);
 
 sub new {
-        my ($class, $v2ibx) = @_;
-        my $self = $class->SUPER::new($v2ibx, 1, 'all');
+        my ($class, $v2writable) = @_;
+        my $self = $class->SUPER::new($v2writable->{-inbox}, 1, 'all');
         # create the DB:
         $self->_xdb_acquire;
         $self->_xdb_release;
@@ -20,6 +20,8 @@ sub new {
         my $pid = fork;
         defined $pid or die "fork failed: $!\n";
         if ($pid == 0) {
+                $v2writable->atfork_child;
+                $v2writable = undef;
                 close $w;
                 eval { thread_worker_loop($self, $r) };
                 die "thread worker died: $@\n" if $@;
diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm
index 29ed23ca..3451261e 100644
--- a/lib/PublicInbox/V2Writable.pm
+++ b/lib/PublicInbox/V2Writable.pm
@@ -84,9 +84,9 @@ sub idx_part {
 sub idx_init {
         my ($self) = @_;
         return if $self->{idx_parts};
-        # first time initialization:
-        my $all = $self->{all} =
-                PublicInbox::SearchIdxThread->new($self->{-inbox});
+
+        # first time initialization, first we create the threader pipe:
+        my $all = $self->{all} = PublicInbox::SearchIdxThread->new($self);
 
         # need to create all parts before initializing msgmap FD
         my $max = $self->{partitions} - 1;
@@ -94,6 +94,8 @@ sub idx_init {
         for my $i (0..$max) {
                 push @$idx, PublicInbox::SearchIdxPart->new($self, $i, $all);
         }
+
+        # Now that all subprocesses are up, we can open the FD for SQLite:
         $all->_msgmap_init->{dbh}->begin_work;
 }
 
@@ -242,4 +244,14 @@ sub lookup_content {
         undef # TODO
 }
 
+sub atfork_child {
+        my ($self) = @_;
+        if (my $parts = $self->{idx_parts}) {
+                $_->atfork_child foreach @$parts;
+        }
+        if (my $im = $self->{im}) {
+                $im->atfork_child;
+        }
+}
+
 1;