lei: q: lock stdout on overview output

[public-inbox.git] / lib / PublicInbox / LeiXSearch.pm

diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index c030b2b29b6f8784a9e733d25192532429f62de8..ee93e074b1cacca62055cff1d507399ea5c224c5 100644 (file)
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -8,7 +8,6 @@ package PublicInbox::LeiXSearch;
 use strict;
 use v5.10.1;
 use parent qw(PublicInbox::LeiSearch PublicInbox::IPC);
-use Sys::Syslog qw(syslog);
 
 sub new {
        my ($class) = @_;
@@ -92,7 +91,9 @@ sub _mset_more ($$) {
 
 sub query_thread_mset { # for --thread
        my ($self, $lei, $ibxish) = @_;
-       local %SIG = (%SIG, $lei->atfork_child_wq($self));
+       my %sig = $lei->atfork_child_wq($self);
+       local @SIG{keys %sig} = values %sig;
+
        my ($srch, $over) = ($ibxish->search, $ibxish->over);
        unless ($srch && $over) {
                my $desc = $ibxish->{inboxdir} // $ibxish->{topdir};
@@ -102,6 +103,8 @@ sub query_thread_mset { # for --thread
        my $mo = { %{$lei->{mset_opt}} };
        my $mset;
        my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei);
+       my $dd = $lei->{dd};
+       $dd->prepare_dedupe;
        do {
                $mset = $srch->mset($mo->{qstr}, $mo);
                my $ids = $srch->mset_to_artnums($mset, $mo);
@@ -111,7 +114,7 @@ sub query_thread_mset { # for --thread
                while ($over->expand_thread($ctx)) {
                        for my $n (@{$ctx->{xids}}) {
                                my $smsg = $over->get_art($n) or next;
-                               # next if $dd->is_smsg_dup($smsg); TODO
+                               next if $dd->is_smsg_dup($smsg);
                                my $mitem = delete $n2item{$smsg->{num}};
                                $each_smsg->($smsg, $mitem);
                                # $self->out($buf .= $ORS);
@@ -125,16 +128,19 @@ sub query_thread_mset { # for --thread
 
 sub query_mset { # non-parallel for non-"--thread" users
        my ($self, $lei, $srcs) = @_;
+       my %sig = $lei->atfork_child_wq($self);
+       local @SIG{keys %sig} = values %sig;
        my $mo = { %{$lei->{mset_opt}} };
        my $mset;
-       local %SIG = (%SIG, $lei->atfork_child_wq($self));
        $self->attach_external($_) for @$srcs;
        my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei);
+       my $dd = $lei->{dd};
+       $dd->prepare_dedupe;
        do {
                $mset = $self->mset($mo->{qstr}, $mo);
                for my $it ($mset->items) {
                        my $smsg = smsg_for($self, $it) or next;
-                       # next if $dd->is_smsg_dup($smsg);
+                       next if $dd->is_smsg_dup($smsg);
                        $each_smsg->($smsg, $it);
                        # $self->out($buf .= $ORS) if defined $buf;
                        #$emit_cb->($smsg);
@@ -143,32 +149,46 @@ sub query_mset { # non-parallel for non-"--thread" users
        $lei->{ovv}->ovv_atexit_child($lei);
 }
 
+sub query_done { # PublicInbox::EOFpipe callback
+       my ($lei) = @_;
+       $lei->{ovv}->ovv_end($lei);
+       $lei->dclose;
+}
+
 sub do_query {
-       my ($self, $lei_orig, $qry_done, $srcs) = @_;
+       my ($self, $lei_orig, $srcs) = @_;
        my ($lei, @io) = $lei_orig->atfork_parent_wq($self);
+       my $remotes = $self->{remotes} // [];
+       pipe(my ($eof_wait, $qry_done)) or die "pipe $!";
        $io[0] = $qry_done; # don't need stdin
-       $io[1]->autoflush(1);
-       $io[2]->autoflush(1);
+
        if ($lei->{opt}->{thread}) {
+               $lei->{-parallel} = scalar(@$remotes) + scalar(@$srcs) - 1;
                for my $ibxish (@$srcs) {
                        $self->wq_do('query_thread_mset', \@io, $lei, $ibxish);
                }
        } else {
+               $lei->{-parallel} = scalar(@$remotes);
                $self->wq_do('query_mset', \@io, $lei, $srcs);
        }
        # TODO
-       for my $rmt (@{$self->{remotes} // []}) {
+       for my $rmt (@$remotes) {
                $self->wq_do('query_thread_mbox', \@io, $lei, $rmt);
        }
-
-       # sent off to children, they will drop remaining references to it
-       close $qry_done;
-}
-
-sub ipc_atfork_child {
-       my ($self) = @_;
-       $SIG{__WARN__} = sub { syslog('warning', "@_") };
-       $self->SUPER::ipc_atfork_child; # PublicInbox::IPC
+       @io = ();
+       close $qry_done; # fully closed when children are done
+
+       # query_done will run when query_*mset close $qry_done
+       if ($lei_orig->{sock}) { # watch for client premature exit
+               require PublicInbox::EOFpipe;
+               PublicInbox::EOFpipe->new($eof_wait, \&query_done, $lei_orig);
+               $lei_orig->{lxs} = $self;
+               $lei_orig->event_step_init;
+       } else {
+               $self->wq_close;
+               read($eof_wait, my $buf, 1); # wait for close($lei->{0})
+               query_done($lei_orig); # may SIGPIPE
+       }
 }
 
 sub ipc_atfork_prepare {