]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/SearchIdxShard.pm
ds: inline set_cloexec
[public-inbox.git] / lib / PublicInbox / SearchIdxShard.pm
1 # Copyright (C) 2018-2021 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3
4 # Internal interface for a single Xapian shard in V2 inboxes.
5 # See L<public-inbox-v2-format(5)> for more info on how we shard Xapian
6 package PublicInbox::SearchIdxShard;
7 use strict;
8 use v5.10.1;
9 use parent qw(PublicInbox::SearchIdx PublicInbox::IPC);
10 use PublicInbox::OnDestroy;
11
12 sub new {
13         my ($class, $v2w, $shard) = @_; # v2w may be ExtSearchIdx
14         my $ibx = $v2w->{ibx};
15         my $self = $ibx ? $class->SUPER::new($ibx, 1, $shard)
16                         : $class->eidx_shard_new($v2w, $shard);
17         # create the DB before forking:
18         $self->idx_acquire;
19         $self->set_metadata_once;
20         $self->idx_release;
21         if ($v2w->{parallel}) {
22                 local $self->{-v2w_afc} = $v2w;
23                 $self->ipc_worker_spawn("shard[$shard]");
24                 # F_SETPIPE_SZ = 1031 on Linux; increasing the pipe size for
25                 # inputs speeds V2Writable batch imports across 8 cores by
26                 # nearly 20%.  Since any of our responses are small, make
27                 # the response pipe as small as possible
28                 if ($^O eq 'linux') {
29                         fcntl($self->{-ipc_req}, 1031, 1048576);
30                         fcntl($self->{-ipc_res}, 1031, 4096);
31                 }
32         }
33         $self;
34 }
35
36 sub _worker_done {
37         my ($self) = @_;
38         if ($self->need_xapian) {
39                 die "$$ $0 xdb not released\n" if $self->{xdb};
40         }
41         die "$$ $0 still in transaction\n" if $self->{txn};
42 }
43
44 sub ipc_atfork_child { # called automatically before ipc_worker_loop
45         my ($self) = @_;
46         my $v2w = delete $self->{-v2w_afc} or die 'BUG: {-v2w_afc} missing';
47         $v2w->atfork_child; # calls ipc_sibling_atfork_child on our siblings
48         $v2w->{current_info} = "[$self->{shard}]"; # for $SIG{__WARN__}
49         $self->begin_txn_lazy;
50         # caller must capture this:
51         PublicInbox::OnDestroy->new($$, \&_worker_done, $self);
52 }
53
54 sub index_eml {
55         my ($self, $eml, $smsg, $eidx_key) = @_;
56         $smsg->{eidx_key} = $eidx_key if defined $eidx_key;
57         $self->ipc_do('add_xapian', $eml, $smsg);
58 }
59
60 # wait for return to determine when ipc_do('commit_txn_lazy') is done
61 sub echo {
62         shift;
63         "@_";
64 }
65
66 sub idx_close {
67         my ($self) = @_;
68         die "transaction in progress $self\n" if $self->{txn};
69         $self->idx_release if $self->{xdb};
70 }
71
72 sub shard_close {
73         my ($self) = @_;
74         $self->ipc_do('idx_close');
75         $self->ipc_worker_stop;
76 }
77
78 sub shard_over_check {
79         my ($self, $over) = @_;
80         if ($self->{-ipc_req} && $over->{dbh}) {
81                 # can't send DB handles over IPC, and use read-only to avoid
82                 # create_tables lock conflict:
83                 $over = PublicInbox::Over->new($over->{dbh}->sqlite_db_filename)
84         }
85         $self->ipc_do('over_check', $over);
86 }
87
88 1;