X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FSearchIdxShard.pm;h=45240e0793a8867e272005d386712e81004af803;hb=af0b0fb7a454470a32c452119d0392e0dedb3fe1;hp=21e81b166b8b29194fdbe354ce3afb6a1d6ea34a;hpb=99d6d24847c499387ce65b09305deebfd51f8930;p=public-inbox.git diff --git a/lib/PublicInbox/SearchIdxShard.pm b/lib/PublicInbox/SearchIdxShard.pm index 21e81b16..45240e07 100644 --- a/lib/PublicInbox/SearchIdxShard.pm +++ b/lib/PublicInbox/SearchIdxShard.pm @@ -1,21 +1,27 @@ -# Copyright (C) 2018-2020 all contributors +# Copyright (C) 2018-2021 all contributors # License: AGPL-3.0+ -# used to interface with a single Xapian shard in V2 repos. +# Internal interface for a single Xapian shard in V2 inboxes. # See L for more info on how we shard Xapian package PublicInbox::SearchIdxShard; use strict; -use warnings; -use base qw(PublicInbox::SearchIdx); +use v5.10.1; +use parent qw(PublicInbox::SearchIdx); +use bytes qw(length); use IO::Handle (); # autoflush +use PublicInbox::Eml; +use PublicInbox::Sigfd; sub new { - my ($class, $v2writable, $shard) = @_; - my $self = $class->SUPER::new($v2writable->{-inbox}, 1, $shard); + my ($class, $v2w, $shard) = @_; # v2w may be ExtSearchIdx + my $ibx = $v2w->{ibx}; + my $self = $ibx ? $class->SUPER::new($ibx, 1, $shard) + : $class->eidx_shard_new($v2w, $shard); # create the DB before forking: - $self->_xdb_acquire; - $self->_xdb_release; - $self->spawn_worker($v2writable, $shard) if $v2writable->{parallel}; + $self->idx_acquire; + $self->set_metadata_once; + $self->idx_release; + $self->spawn_worker($v2w, $shard) if $v2w->{parallel}; $self; } @@ -23,12 +29,15 @@ sub spawn_worker { my ($self, $v2w, $shard) = @_; my ($r, $w); pipe($r, $w) or die "pipe failed: $!\n"; - binmode $r, ':raw'; - binmode $w, ':raw'; $w->autoflush(1); + my $oldset = PublicInbox::Sigfd::block_signals(); my $pid = fork; defined $pid or die "fork failed: $!\n"; if ($pid == 0) { + eval { PublicInbox::DS->Reset }; + # these signals are localized in parent + $SIG{$_} = 'IGNORE' for (qw(TERM INT QUIT)); + PublicInbox::Sigfd::sig_setmask($oldset); my $bnote = $v2w->atfork_child; close $w or die "failed to close: $!"; @@ -41,66 +50,125 @@ sub spawn_worker { die "unexpected MM $self->{mm}" if $self->{mm}; exit; } + PublicInbox::Sigfd::sig_setmask($oldset); $self->{pid} = $pid; $self->{w} = $w; close $r or die "failed to close: $!"; } +sub eml ($$) { + my ($r, $len) = @_; + return if $len == 0; + my $n = read($r, my $bref, $len) or die "read: $!\n"; + $n == $len or die "short read: $n != $len\n"; + PublicInbox::Eml->new(\$bref); +} + +# this reads all the writes to $self->{w} from the parent process sub shard_worker_loop ($$$$$) { my ($self, $v2w, $r, $shard, $bnote) = @_; - $0 = "pi-v2-shard[$shard]"; + $0 = "shard[$shard]"; $self->begin_txn_lazy; - while (my $line = $r->getline) { + while (my $line = readline($r)) { + chomp $line; $v2w->{current_info} = "[$shard] $line"; - if ($line eq "commit\n") { + if ($line eq 'commit') { $self->commit_txn_lazy; - } elsif ($line eq "close\n") { - $self->_xdb_release; - } elsif ($line eq "barrier\n") { + } elsif ($line eq 'close') { + $self->idx_release; + } elsif ($line eq 'barrier') { $self->commit_txn_lazy; # no need to lock < 512 bytes is atomic under POSIX print $bnote "barrier $shard\n" or die "write failed for barrier $!\n"; - } elsif ($line =~ /\AD ([a-f0-9]{40,}) (.+)\n\z/s) { - my ($oid, $mid) = ($1, $2); - $self->begin_txn_lazy; - $self->remove_by_oid($oid, $mid); + } elsif ($line =~ /\AD ([0-9]+)\z/s) { + $self->remove_by_docid($1 + 0); + } elsif ($line =~ s/\A\+X //) { + my ($len, $docid, $eidx_key) = split(/ /, $line, 3); + $self->add_eidx_info($docid, $eidx_key, eml($r, $len)); + } elsif ($line =~ s/\A-X //) { + my ($len, $docid, $eidx_key) = split(/ /, $line, 3); + $self->remove_eidx_info($docid, $eidx_key, + eml($r, $len)); + } elsif ($line =~ s/\A=K (\d+) //) { + $self->set_keywords($1 + 0, split(/ /, $line)); + } elsif ($line =~ s/\A-K (\d+) //) { + $self->remove_keywords($1 + 0, split(/ /, $line)); + } elsif ($line =~ s/\A\+K (\d+) //) { + $self->add_keywords($1 + 0, split(/ /, $line)); + } elsif ($line =~ s/\AO ([^\n]+)//) { + my $over_fn = $1; + $over_fn =~ tr/\0/\n/; + $self->over_check(PublicInbox::Over->new($over_fn)); } else { - chomp $line; - my ($bytes, $num, $blob, $mid, $autime, $cotime) = - split(/ /, $line); + my $eidx_key; + if ($line =~ s/\AX=(.+)\0//) { + $eidx_key = $1; + $v2w->{current_info} =~ s/\0/\\0 /; + } + # n.b. $mid may contain spaces(!) + my ($len, $bytes, $num, $oid, $ds, $ts, $tid, $mid) + = split(/ /, $line, 8); $self->begin_txn_lazy; - my $n = read($r, my $msg, $bytes) or die "read: $!\n"; - $n == $bytes or die "short read: $n != $bytes\n"; - my $mime = PublicInbox::MIME->new(\$msg); - $self->{autime} = $autime; - $self->{cotime} = $cotime; my $smsg = bless { bytes => $bytes, num => $num + 0, - blob => $blob, + blob => $oid, mid => $mid, + tid => $tid, + ds => $ds, + ts => $ts, }, 'PublicInbox::Smsg'; - $self->add_message($mime, $smsg); + $smsg->{eidx_key} = $eidx_key if defined($eidx_key); + $self->add_message(eml($r, $len), $smsg); } } $self->worker_done; } -# called by V2Writable sub index_raw { - my ($self, $msgref, $mime, $smsg, $times) = @_; - my $at = $times->{autime} // time; - my $ct = $times->{cotime} // time; + my ($self, $msgref, $eml, $smsg, $eidx_key) = @_; if (my $w = $self->{w}) { - print $w join(' ', @$smsg{qw(bytes num blob mid)}, $at, $ct), + my @ekey = defined($eidx_key) ? ("X=$eidx_key\0") : (); + $msgref //= \($eml->as_string); + $smsg->{raw_bytes} //= length($$msgref); + # mid must be last, it can contain spaces (but not LF) + print $w @ekey, join(' ', @$smsg{qw(raw_bytes bytes + num blob ds ts tid mid)}), "\n", $$msgref or die "failed to write shard $!\n"; } else { - $$msgref = undef; + if ($eml) { + undef($$msgref) if $msgref; + } else { # --xapian-only + --sequential-shard: + $eml = PublicInbox::Eml->new($msgref); + } $self->begin_txn_lazy; - $self->{autime} = $at; - $self->{cotime} = $ct; - $self->add_message($mime, $smsg); + $smsg->{eidx_key} = $eidx_key if defined $eidx_key; + $self->add_message($eml, $smsg); + } +} + +sub shard_add_eidx_info { + my ($self, $docid, $eidx_key, $eml) = @_; + if (my $w = $self->{w}) { + my $hdr = $eml->header_obj->as_string; + my $len = length($hdr); + print $w "+X $len $docid $eidx_key\n", $hdr or + die "failed to write shard: $!"; + } else { + $self->add_eidx_info($docid, $eidx_key, $eml); + } +} + +sub shard_remove_eidx_info { + my ($self, $docid, $eidx_key, $eml) = @_; + if (my $w = $self->{w}) { + my $hdr = $eml ? $eml->header_obj->as_string : ''; + my $len = length($hdr); + print $w "-X $len $docid $eidx_key\n", $hdr or + die "failed to write shard: $!"; + } else { + $self->remove_eidx_info($docid, $eidx_key, $eml); } } @@ -108,8 +176,7 @@ sub atfork_child { close $_[0]->{w} or die "failed to close write pipe: $!\n"; } -# called by V2Writable: -sub remote_barrier { +sub shard_barrier { my ($self) = @_; if (my $w = $self->{w}) { print $w "barrier\n" or die "failed to print: $!"; @@ -118,4 +185,74 @@ sub remote_barrier { } } +sub shard_commit { + my ($self) = @_; + if (my $w = $self->{w}) { + print $w "commit\n" or die "failed to write commit: $!"; + } else { + $self->commit_txn_lazy; + } +} + +sub shard_close { + my ($self) = @_; + if (my $w = delete $self->{w}) { + my $pid = delete $self->{pid} or die "no process to wait on\n"; + print $w "close\n" or die "failed to write to pid:$pid: $!\n"; + close $w or die "failed to close pipe for pid:$pid: $!\n"; + waitpid($pid, 0) == $pid or die "remote process did not finish"; + $? == 0 or die ref($self)." pid:$pid exited with: $?"; + } else { + die "transaction in progress $self\n" if $self->{txn}; + $self->idx_release if $self->{xdb}; + } +} + +sub shard_remove { + my ($self, $num) = @_; + if (my $w = $self->{w}) { # triggers remove_by_docid in a shard child + print $w "D $num\n" or die "failed to write remove $!"; + } else { # same process + $self->remove_by_docid($num); + } +} + +sub shard_set_keywords { + my ($self, $docid, @kw) = @_; + if (my $w = $self->{w}) { # triggers remove_by_docid in a shard child + print $w "=K $docid @kw\n" or die "failed to write: $!"; + } else { # same process + $self->set_keywords($docid, @kw); + } +} + +sub shard_remove_keywords { + my ($self, $docid, @kw) = @_; + if (my $w = $self->{w}) { # triggers remove_by_docid in a shard child + print $w "-K $docid @kw\n" or die "failed to write: $!"; + } else { # same process + $self->remove_keywords($docid, @kw); + } +} + +sub shard_add_keywords { + my ($self, $docid, @kw) = @_; + if (my $w = $self->{w}) { # triggers remove_by_docid in a shard child + print $w "+K $docid @kw\n" or die "failed to write: $!"; + } else { # same process + $self->add_keywords($docid, @kw); + } +} + +sub shard_over_check { + my ($self, $over) = @_; + if (my $w = $self->{w}) { # triggers remove_by_docid in a shard child + my ($over_fn) = $over->{dbh}->sqlite_db_filename; + $over_fn =~ tr/\n/\0/; + print $w "O $over_fn\n" or die "failed to write over $!"; + } else { + $self->over_check($over); + } +} + 1;