lei: retry_reopen on read-only Xapian access

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

diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index b3fd79d005c9ad00954c6d6e9b5edfce4e653c3d..2e548a7a9f4574b4c4a74791f295d3426b23abb2 100644 (file)
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -67,17 +67,23 @@ sub remotes { @{$_[0]->{remotes} // []} }
 # called by PublicInbox::Search::xdb (usually via ->mset)
 sub xdb_shards_flat { @{$_[0]->{shards_flat} // []} }
 
-sub mitem_kw ($$;$) {
-       my ($smsg, $mitem, $flagged) = @_;
-       my $kw = xap_terms('K', my $doc = $mitem->get_document);
+sub _mitem_kw { # retry_reopen callback
+       my ($srch, $smsg, $mitem, $flagged) = @_;
+       my $doc = $mitem->get_document;
+       my $kw = xap_terms('K', $doc);
        $kw->{flagged} = 1 if $flagged;
+       my $L = xap_terms('L', $doc);
        # we keep the empty {kw} array here to prevent expensive work in
        # ->xsmsg_vmd, _unbless_smsg will clobber it iff it's empty
        $smsg->{kw} = [ sort keys %$kw ];
-       my $L = xap_terms('L', $doc);
        $smsg->{L} = [ sort keys %$L ] if scalar(keys %$L);
 }
 
+sub mitem_kw ($$$;$) {
+       my ($srch, $smsg, $mitem, $flagged) = @_;
+       $srch->retry_reopen(\&_mitem_kw, $smsg, $mitem, $flagged);
+}
+
 # like over->get_art
 sub smsg_for {
        my ($self, $mitem) = @_;
@@ -90,7 +96,7 @@ sub smsg_for {
        my $smsg = $ibx->over->get_art($num);
        return if $smsg->{bytes} == 0; # external message
        if ($ibx->can('msg_keywords')) {
-               mitem_kw($smsg, $mitem);
+               mitem_kw($self, $smsg, $mitem);
        }
        $smsg;
 }
@@ -139,11 +145,16 @@ sub mset_progress {
                pkt_do($lei->{pkt_op_p}, 'mset_progress', @_);
        } else { # single lei-daemon consumer
                my ($desc, $mset_size, $mset_total_est) = @_;
-               $lei->{-mset_total} += $mset_size;
+               $lei->{-mset_total} += $mset_size if $mset_total_est ne '?';
                $lei->qerr("# $desc $mset_size/$mset_total_est");
        }
 }
 
+sub l2m_progress {
+       my ($lei, $nr) = @_;
+       $lei->{-nr_write} += $nr;
+}
+
 sub query_one_mset { # for --threads and l2m w/o sort
        my ($self, $ibxish) = @_;
        local $0 = "$0 query_one_mset";
@@ -189,7 +200,8 @@ sub query_one_mset { # for --threads and l2m w/o sort
                                        my $mitem = delete $n2item{$n};
                                        next if $smsg->{bytes} == 0;
                                        if ($mitem && $can_kw) {
-                                               mitem_kw($smsg, $mitem, $fl);
+                                               mitem_kw($srch, $smsg, $mitem,
+                                                       $fl);
                                        } elsif ($mitem && $fl) {
                                                # call ->xsmsg_vmd, later
                                                $smsg->{lei_q_tt_flagged} = 1;
@@ -205,7 +217,7 @@ sub query_one_mset { # for --threads and l2m w/o sort
                                my $mitem = $items[$i++];
                                my $smsg = $over->get_art($n) or next;
                                next if $smsg->{bytes} == 0;
-                               mitem_kw($smsg, $mitem, $fl) if $can_kw;
+                               mitem_kw($srch, $smsg, $mitem, $fl) if $can_kw;
                                $each_smsg->($smsg, $mitem);
                        }
                }
@@ -233,7 +245,7 @@ sub query_combined_mset { # non-parallel for non-"--threads" users
        do {
                $mset = $self->mset($mo->{qstr}, $mo);
                mset_progress($lei, 'xsearch', $mset->size,
-                               $mset->size, $mset->get_matches_estimated);
+                               $mset->get_matches_estimated);
                wait_startq($lei); # wait for keyword updates
                for my $mitem ($mset->items) {
                        my $smsg = smsg_for($self, $mitem) or next;
@@ -352,11 +364,13 @@ sub query_done { # EOF callback for main daemon
        if (my $lxs = delete $lei->{lxs}) {
                $lxs->wq_wait_old(\&xsearch_done_wait, $lei);
        }
+       if ($lei->{opt}->{'mail-sync'} && !$lei->{sto}) {
+               warn "BUG: {sto} missing with --mail-sync";
+       }
        my $wait = $lei->{sto} ? $lei->{sto}->ipc_do('done') : undef;
        $lei->{ovv}->ovv_end($lei);
-       my @out;
+       my $start_mua;
        if ($l2m) { # close() calls LeiToMail reap_compress
-               @out = (" in $lei->{ovv}->{dst}");
                if (my $out = delete $lei->{old_1}) {
                        if (my $mbout = $lei->{1}) {
                                close($mbout) or return $lei->fail(<<"");
@@ -370,17 +384,24 @@ Error closing $lei->{ovv}->{dst}: $!
                        $lei->poke_mua;
                } else { # mbox users
                        delete $l2m->{mbl}; # drop dotlock
-                       $lei->start_mua;
+                       $start_mua = 1;
                }
        }
-       $lei->{-progress} and
-               $lei->err('# ', $lei->{-mset_total} // 0, " matches", @out);
+       if ($lei->{-progress}) {
+               my $tot = $lei->{-mset_total} // 0;
+               my $nr = $lei->{-nr_write} // 0;
+               $lei->qerr($l2m ?
+                       "# $nr written to $lei->{ovv}->{dst} ($tot matches)" :
+                       "# $tot matches");
+       }
+       $lei->start_mua if $start_mua;
        $lei->dclose;
 }
 
 sub do_post_augment {
        my ($lei) = @_;
        my $l2m = $lei->{l2m} or return; # client disconnected
+       $lei->fchdir or return;
        my $err;
        eval { $l2m->post_augment($lei) };
        $err = $@;
@@ -456,6 +477,7 @@ sub do_query {
                '+' => [ \&incr_post_augment, $lei ],
                '' => [ \&query_done, $lei ],
                'mset_progress' => [ \&mset_progress, $lei ],
+               'l2m_progress' => [ \&l2m_progress, $lei ],
                'x_it' => [ $lei->can('x_it'), $lei ],
                'child_error' => [ $lei->can('child_error'), $lei ],
                'incr_start_query' => [ \&incr_start_query, $self, $l2m ],
@@ -471,11 +493,22 @@ sub do_query {
                if ($lei->{opt}->{augment} && delete $lei->{early_mua}) {
                        $lei->start_mua;
                }
+               my $F_SETPIPE_SZ = $^O eq 'linux' ? 1031 : undef;
+               if ($l2m->{-wq_nr_workers} > 1 &&
+                               $l2m->{base_type} =~ /\A(?:maildir|mbox)\z/) {
+                       # setup two barriers to coordinate dedupe_nr
+                       # between l2m workers
+                       pipe(my ($a_r, $a_w)) or die "pipe: $!";
+                       fcntl($a_r, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
+                       pipe(my ($b_r, $b_w)) or die "pipe: $!";
+                       fcntl($b_r, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
+                       $l2m->{au_peers} = [ $a_r, $a_w, $b_r, $b_w ];
+               }
                $l2m->wq_workers_start('lei2mail', undef,
                                        $lei->oldset, { lei => $lei });
                pipe($lei->{startq}, $lei->{au_done}) or die "pipe: $!";
-               # 1031: F_SETPIPE_SZ
-               fcntl($lei->{startq}, 1031, 4096) if $^O eq 'linux';
+               fcntl($lei->{startq}, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
+               delete $l2m->{au_peers};
        }
        $self->wq_workers_start('lei_xsearch', undef,
                                $lei->oldset, { lei => $lei });
@@ -485,7 +518,7 @@ sub do_query {
        $self->{opt_threads} = $lei->{opt}->{threads};
        $self->{opt_sort} = $lei->{opt}->{'sort'};
        if ($l2m) {
-               $l2m->net_merge_complete unless $lei->{auth};
+               $l2m->net_merge_all_done unless $lei->{auth};
        } else {
                start_query($self);
        }