We only need to write one byte at MUA start instead of a byte
for every LeiXSearch worker. Also, make sure it succeeds by
enabling autodie for syswrite.
When reading, we can rely on `:perlio' layer `read' semantics
to retry on EINTR to avoid looping and other error checking.
---
lib/PublicInbox/LEI.pm | 9 +++++----
lib/PublicInbox/LeiXSearch.pm | 28 +++++++++-------------------
2 files changed, 14 insertions(+), 23 deletions(-)
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 3ccdd4f7..56e4c001 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -9,7 +9,7 @@ package PublicInbox::LEI;
use v5.12;
use parent qw(PublicInbox::DS PublicInbox::LeiExternal
PublicInbox::LeiQuery);
-use autodie qw(bind chdir fork open socket socketpair unlink);
+use autodie qw(bind chdir fork open socket socketpair syswrite unlink);
use Getopt::Long ();
use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un);
use Errno qw(EPIPE EAGAIN ECONNREFUSED ENOENT ECONNRESET);
@@ -1031,9 +1031,10 @@ sub start_mua {
$io->[0] = $self->{1} if $self->{opt}->{stdin} && -t $self->{1};
send_exec_cmd($self, $io, \@cmd, {});
}
- if ($self->{lxs} && $self->{au_done}) { # kick wait_startq
- syswrite($self->{au_done}, 'q' x ($self->{lxs}->{jobs} // 0));
- }
+
+ # kick wait_startq:
+ syswrite($self->{au_done}, 'q') if $self->{lxs} && $self->{au_done};
+
return unless -t $self->{2}; # XXX how to determine non-TUI MUAs?
$self->{opt}->{quiet} = 1;
delete $self->{-progress};
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 25b66b3b..241b9dab 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -122,26 +122,16 @@ sub _mset_more ($$) {
$size >= $mo->{limit} && (($mo->{offset} += $size) < $mo->{total});
}
-# $startq will EOF when do_augment is done augmenting and allow
+# $startq will see `q' in do_post_augment -> start_mua if spawning MUA.
+# Otherwise $startq will EOF when do_augment is done augmenting and allow
# query_combined_mset and query_thread_mset to proceed.
sub wait_startq ($) {
my ($lei) = @_;
- my $startq = delete $lei->{startq} or return;
- while (1) {
- my $n = sysread($startq, my $do_augment_done, 1);
- if (defined $n) {
- return if $n == 0; # no MUA
- if ($do_augment_done eq 'q') {
- $lei->{opt}->{quiet} = 1;
- delete $lei->{opt}->{verbose};
- delete $lei->{-progress};
- } else {
- die "BUG: do_augment_done=`$do_augment_done'";
- }
- return;
- }
- die "wait_startq: $!" unless $!{EINTR};
- }
+ read(delete($lei->{startq}) // return, my $buf, 1) or return; # EOF
+ die "BUG: wrote `$buf' to au_done" if $buf ne 'q';
+ $lei->{opt}->{quiet} = 1;
+ delete $lei->{opt}->{verbose};
+ delete $lei->{-progress};
}
sub mset_progress {
@@ -451,10 +441,10 @@ sub do_post_augment {
$lei->fail("$err");
}
if (!$err && delete $lei->{early_mua}) { # non-augment case
- eval { $lei->start_mua };
+ eval { $lei->start_mua }; # may trigger wait_startq
$lei->fail($@) if $@;
}
- close(delete $lei->{au_done}); # triggers wait_startq in lei_xsearch
+ close(delete $lei->{au_done}); # trigger wait_startq if start_mua didn't
}
sub incr_post_augment { # called whenever an l2m shard finishes augment
