From: Eric Wong Date: Fri, 3 Jul 2026 12:23:40 +0000 (+0000) Subject: lei reindex: avoid forking worker process X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;p=thirdparty%2Fpublic-inbox.git 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 a09d1a19d..43e2870bf 100644 --- 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); + $self->{sock} // return; $self->{epwatch} //= do { $self->SUPER::new($self->{sock}, EPOLLIN); \undef; diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm index bf95eaa73..1d32967b8 100644 --- 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 } +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) = @_; @@ -1208,6 +1223,7 @@ sub event_step { if ($buf eq '') { _drop_wq($self); # EOF, client disconnected dclose($self); + _drop_sto_long($self); $buf = 'TERM'; } if ($buf =~ /\A(?:STOP|CONT|TERM)\z/) { diff --git a/lib/PublicInbox/LeiReindex.pm b/lib/PublicInbox/LeiReindex.pm index 3f109f33a..644fe71b0 100644 --- 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; -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); - 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; diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm index 960cf8d51..72f9c42c5 100644 --- a/lib/PublicInbox/LeiStore.pm +++ b/lib/PublicInbox/LeiStore.pm @@ -403,13 +403,55 @@ sub reindex_art { \&_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); - $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 } +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 diff --git a/t/lei-reindex.t b/t/lei-reindex.t index 73346ee8f..3761ae001 100644 --- a/t/lei-reindex.t +++ b/t/lei-reindex.t @@ -3,10 +3,13 @@ # License: AGPL-3.0+ 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'); - 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;