lei reindex: avoid forking worker process

Instead of spawning a potentially expensive dedicated process to
feed numeric docids to the lei/store process, we can instead
rely on requeue + event_step and the DS event loop of the
lei/store working to iterate through the query list.

In addition to saving memory from forking, handling iteration
entirely within one process saves some syscalls and IPC
overhead.

We'll also flesh out lei-reindex.t a bit to test reindex
explicitly, even if lei-index.t already tests it.

diff --git a/lib/PublicInbox/Git.pm b/lib/PublicInbox/Git.pm
index a09d1a19d51c139e0918cd029e2ff99fb2b88641..43e2870bf0cbcd75a38e2288bb04674fcfd47d53 100644 (file)
--- a/lib/PublicInbox/Git.pm
+++ b/lib/PublicInbox/Git.pm
@@ -677,6 +677,7 @@ sub schedule_cleanup {
 sub watch_async ($) {
        my ($self) = @_;
        schedule_cleanup($self);
 sub watch_async ($) {
        my ($self) = @_;
        schedule_cleanup($self);

+       $self->{sock} // return;

        $self->{epwatch} //= do {
                $self->SUPER::new($self->{sock}, EPOLLIN);
                \undef;
        $self->{epwatch} //= do {
                $self->SUPER::new($self->{sock}, EPOLLIN);
                \undef;

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index bf95eaa738e8f2011d1c7f94901d09da169888a7..1d32967b85d2c8787a406baa7da19753535e8c8c 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -1192,6 +1192,21 @@ sub dclose {
        $self->close if $self->{-event_init_done}; # PublicInbox::DS::close
 }
 
        $self->close if $self->{-event_init_done}; # PublicInbox::DS::close
 }
 

+sub _drop_sto_long {
+       my ($self) = @_;
+       my $reqid = delete $self->{-long_reqid} or return;
+       ($self->{sto} // return)->wq_do('abort_long_req', $reqid);
+}
+
+sub long_reqid {
+       my ($self) = @_;
+       $self->{-long_reqid} // do {
+               my @st = stat $self->{sock} or die "stat(lei->{sock}): $!";
+               # n.b. inode may be recycled by kernel, so prefix with time
+               $self->{-long_reqid} = PublicInbox::DS::now.".$st[1]";
+       };
+}
+

 # for long-running results
 sub event_step {
        my ($self) = @_;
 # for long-running results
 sub event_step {
        my ($self) = @_;


@@ -1208,6 +1223,7 @@ sub event_step {
                if ($buf eq '') {
                        _drop_wq($self); # EOF, client disconnected
                        dclose($self);
                if ($buf eq '') {
                        _drop_wq($self); # EOF, client disconnected
                        dclose($self);

+                       _drop_sto_long($self);

                        $buf = 'TERM';
                }
                if ($buf =~ /\A(?:STOP|CONT|TERM)\z/) {
                        $buf = 'TERM';
                }
                if ($buf =~ /\A(?:STOP|CONT|TERM)\z/) {

diff --git a/lib/PublicInbox/LeiReindex.pm b/lib/PublicInbox/LeiReindex.pm
index 3f109f33a975726aa00c368e1e7199b2bd9a25db..644fe71b0e2ffef8c30ac9c4c36a1e1dc22ca1ac 100644 (file)
--- a/lib/PublicInbox/LeiReindex.pm
+++ b/lib/PublicInbox/LeiReindex.pm
@@ -4,46 +4,21 @@
 # "lei reindex" command to reindex everything in lei/store
 package PublicInbox::LeiReindex;
 use v5.12;
 # "lei reindex" command to reindex everything in lei/store
 package PublicInbox::LeiReindex;
 use v5.12;

-use parent qw(PublicInbox::IPC);
-
-sub reindex_full {
-       my ($lei) = @_;
-       my $sto = $lei->{sto};
-       my $max = $sto->search->over(1)->max;
-       $lei->qerr("# reindexing 1..$max");
-       $sto->wq_do('reindex_art', $_) for (1..$max);
-}
-
-sub reindex_store { # via wq_do
-       my ($self) = @_;
-       my ($lei, $argv) = delete @$self{qw(lei argv)};
-       if (!@$argv) {
-               reindex_full($lei);
-       }
-}
+use autodie qw(pipe);
+use PublicInbox::EOFpipe;

 
 sub lei_reindex {
        my ($lei, @argv) = @_;
        my $sto = $lei->_lei_store or return $lei->fail('nothing indexed');
        $sto->write_prepare($lei);
 
 sub lei_reindex {
        my ($lei, @argv) = @_;
        my $sto = $lei->_lei_store or return $lei->fail('nothing indexed');
        $sto->write_prepare($lei);

-       my $self = bless { lei => $lei, argv => \@argv }, __PACKAGE__;
-       my ($op_c, $ops) = $lei->workers_start($self, 1);
-       $lei->{wq1} = $self;
-       $lei->wait_wq_events($op_c, $ops);
-       $self->wq_do('reindex_store');
-       $self->wq_close;
-}
-
-sub _lei_wq_eof { # EOF callback for main lei daemon
-       my ($lei) = @_;
-       $lei->{sto}->wq_do('reindex_done');
-       $lei->wq_eof;
-}
-
-sub ipc_atfork_child {
-       my ($self) = @_;
-       $self->{lei}->_lei_atfork_child;
-       $self->SUPER::ipc_atfork_child;
+       my $max = $sto->search->over(1)->max;
+       $lei->qerr("# reindexing 1..$max");
+       pipe(my $r, my $w);
+       my @io = (@$lei{2, 'sock'}, $w);
+       PublicInbox::EOFpipe->new($r, $lei->can('dclose'), $lei);
+       $lei->event_step_init; # ensure Ctrl-C can stop reindex
+       # FIXME: wq_io_do can still block:
+       $sto->wq_io_do('reindex_range', \@io, 1, $max, $lei->long_reqid);

 }
 
 1;
 }
 
 1;

diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm
index 960cf8d51263a31f735f07440e6c4a41c9fc8b79..72f9c42c5a4dfd2f991e7eb02bc7f3845d3706d8 100644 (file)
--- a/lib/PublicInbox/LeiStore.pm
+++ b/lib/PublicInbox/LeiStore.pm
@@ -403,13 +403,55 @@ sub reindex_art {
                                \&_reindex_1, $smsg);
 }
 
                                \&_reindex_1, $smsg);
 }
 

-sub reindex_done {
-       my ($self) = @_;
+sub _ridx_done { # OnDestroy cb
+       my ($self, $rireq) = @_;
+       local @$self{0, 1} = @$rireq{qw(errfh lei_sock)};
+       barrier($self);
+       # $rireq->{eof_wr} goes out-of-scope to trigger lei->dclose
+}
+
+sub reindex_done ($$) {
+       my ($self, $rireq) = @_;

        my ($eidx, $tl) = eidx_init($self);
        my ($eidx, $tl) = eidx_init($self);

-       $eidx->git->async_wait_all;
+       $eidx->git->watch_async;
+       $eidx->git->async_barrier(on_destroy(\&_ridx_done, $self, $rireq));

        # ->done to be called via sto_barrier_request
 }
 
        # ->done to be called via sto_barrier_request
 }
 

+sub event_step {
+       my ($self) = @_;
+       for my $reqid (keys %{$self->{ridx}}) {
+               my $rireq = $self->{ridx}->{$reqid};
+               if ($rireq->{cur} < $rireq->{max}) {
+                       reindex_art($self, $rireq->{cur}++);
+               } else {
+                       delete $self->{ridx}->{$reqid};
+                       reindex_done($self, $rireq);
+               }
+       }
+       # run ->event_step again if more to do
+       keys(%{$self->{ridx}}) ? PublicInbox::DS::requeue($self)
+                               : delete($self->{ridx});
+}
+
+sub reindex_range {
+       my ($self, $min, $max, $reqid) = @_;
+       $self->{ridx}->{$reqid} = {
+               errfh => $self->{0},
+               lei_sock => $self->{1},
+               eof_wr => $self->{2}, # calls lei->dclose via EOFpipe
+               cur => $min, max => $max,
+       };
+       PublicInbox::DS::requeue($self); # runs ->event_step
+}
+
+sub abort_long_req {
+       my ($self, $reqid) = @_;
+       my $reqs = $self->{ridx} // return;
+       my $rireq = $reqs->{$reqid} // return;
+       $rireq->{cur} = $rireq->{max}; # ->event_step finishes up
+}
+

 sub add_eml {
        my ($self, $eml, $vmd, $xoids) = @_;
        my $im = $self->{-fake_im} // $self->importer; # may create new epoch
 sub add_eml {
        my ($self, $eml, $vmd, $xoids) = @_;
        my $im = $self->{-fake_im} // $self->importer; # may create new epoch

diff --git a/t/lei-reindex.t b/t/lei-reindex.t
index 73346ee8f08fd2527d593ab2db6bc26d6758d954..3761ae001fb1022e658c9cc1da2a4a86608f51b0 100644 (file)
--- a/t/lei-reindex.t
+++ b/t/lei-reindex.t
@@ -3,10 +3,13 @@
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 use v5.12; use PublicInbox::TestCommon;
 require_mods(qw(lei));
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 use v5.12; use PublicInbox::TestCommon;
 require_mods(qw(lei));

-my ($tmpdir, $for_destroy) = tmpdir;
+

 test_lei(sub {
        ok(!lei('reindex'), 'reindex fails w/o store');
 test_lei(sub {
        ok(!lei('reindex'), 'reindex fails w/o store');

-       like $lei_err, qr/nothing indexed/, "`nothing indexed' noted";
+       like $lei_err, qr/nothing indexed/, "`nothing reindexed' noted";
+
+       lei_ok qw(import t/data/0001.patch);
+       lei_ok 'reindex', \'reindex successful after import';

 });
 
 done_testing;
 });
 
 done_testing;