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.

diff --git a/lib/PublicInbox/Import.pm b/lib/PublicInbox/Import.pm
index b650e4ef1a1dc9d6a610a812cdca9022b71f53d3..ac46c0cbc13968777f232db9516ee07b5e777ca9 100644 (file)
--- 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 5582d672d0de343b19dfc297446056ed8217407c..64e5263657160430ecec3de2fa88d06adec3f155 100644 (file)
--- 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 6471309ee187aa5e1c8b447a70a77722b61d4a93..7df07a65dde31bc1e7f0c0a996968330fda337f0 100644 (file)
--- 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 29ed23caaa76fb87fcce3e48caeb4ec7c7f26a23..3451261e4ee1f902dddce340a4a38780cfae3795 100644 (file)
--- 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;