X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FSearchIdxShard.pm;h=579ed1965962c5993c1488a9442a3a027e8f3da3;hb=0d38f65c490466837ae091afa7a7b6f59d04ce7c;hp=fa77a9f9b96a0780d1484f16a2293df0f4e674b3;hpb=039e1492f773ad1664e3d8a8bac7d67b1de9392f;p=public-inbox.git diff --git a/lib/PublicInbox/SearchIdxShard.pm b/lib/PublicInbox/SearchIdxShard.pm index fa77a9f9..579ed196 100644 --- a/lib/PublicInbox/SearchIdxShard.pm +++ b/lib/PublicInbox/SearchIdxShard.pm @@ -10,11 +10,13 @@ use parent qw(PublicInbox::SearchIdx); use bytes qw(length); use IO::Handle (); # autoflush use PublicInbox::Eml; +use PublicInbox::Sigfd; sub new { - my ($class, $v2w, $shard) = @_; + my ($class, $v2w, $shard) = @_; # v2w may be ExtSearchIdx my $ibx = $v2w->{ibx}; - my $self = $class->SUPER::new($ibx, 1, $shard); + my $self = $ibx ? $class->SUPER::new($ibx, 1, $shard) + : $class->eidx_shard_new($v2w, $shard); # create the DB before forking: $self->idx_acquire; $self->set_metadata_once; @@ -28,9 +30,13 @@ sub spawn_worker { my ($r, $w); pipe($r, $w) or die "pipe failed: $!\n"; $w->autoflush(1); + my $oldset = PublicInbox::Sigfd::block_signals(); my $pid = fork; defined $pid or die "fork failed: $!\n"; if ($pid == 0) { + # 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: $!"; @@ -43,6 +49,7 @@ 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: $!"; @@ -50,6 +57,7 @@ sub spawn_worker { 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); @@ -58,7 +66,7 @@ sub eml ($$) { # 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 = readline($r)) { $v2w->{current_info} = "[$shard] $line"; @@ -71,23 +79,27 @@ sub shard_worker_loop ($$$$$) { # 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,}) ([0-9]+)\n\z/s) { - $self->remove_by_oid($1, $2 + 0); + } elsif ($line =~ /\AD ([0-9]+)\n\z/s) { + $self->remove_by_docid($1 + 0); } elsif ($line =~ s/\A\+X //) { - my ($len, $docid, $xnum, $oid, $eidx_key) = - split(/ /, $line, 5); - $self->add_xref3($docid, $xnum, $oid, $eidx_key, - eml($r, $len)); + my ($len, $docid, $eidx_key) = split(/ /, $line, 3); + chomp $eidx_key; + $self->add_eidx_info($docid, $eidx_key, eml($r, $len)); } elsif ($line =~ s/\A-X //) { - my ($len, $docid, $xnum, $oid, $eidx_key) = - split(/ /, $line, 5); - $self->remove_xref3($docid, $xnum, $oid, - $eidx_key, eml($r, $len)); + my ($len, $docid, $eidx_key) = split(/ /, $line, 3); + chomp $eidx_key; + $self->remove_eidx_info($docid, $eidx_key, + eml($r, $len)); + } elsif ($line =~ s/\AO ([^\n]+)\n//) { + my $over_fn = $1; + $over_fn =~ tr/\0/\n/; + $self->over_check(PublicInbox::Over->new($over_fn)); } else { chomp $line; my $eidx_key; - if ($line =~ s/\AX(.+)\0//) { + 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) @@ -112,14 +124,11 @@ sub shard_worker_loop ($$$$$) { sub index_raw { my ($self, $msgref, $eml, $smsg, $ibx) = @_; if (my $w = $self->{w}) { - if ($ibx) { - print $w 'X', $ibx->eidx_key, "\0" or die - "failed to write shard: $!\n"; - } + my @ekey = $ibx ? ('X='.$ibx->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 join(' ', @$smsg{qw(raw_bytes bytes + 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 { @@ -134,29 +143,27 @@ sub index_raw { } } -sub shard_add_xref3 { - my ($self, $docid, $xnum, $oid, $xibx, $eml) = @_; - my $eidx_key = $xibx->eidx_key; +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 $xnum $oid $eidx_key\n", $hdr or + print $w "+X $len $docid $eidx_key\n", $hdr or die "failed to write shard: $!"; } else { - $self->add_xref3($docid, $xnum, $oid, $eidx_key, $eml); + $self->add_eidx_info($docid, $eidx_key, $eml); } } -sub shard_remove_xref3 { - my ($self, $docid, $oid, $xibx, $eml) = @_; - my $eidx_key = $xibx->eidx_key; +sub shard_remove_eidx_info { + my ($self, $docid, $eidx_key, $eml) = @_; if (my $w = $self->{w}) { - my $hdr = $eml->header_obj->as_string; + my $hdr = $eml ? $eml->header_obj->as_string : ''; my $len = length($hdr); - print $w "-X $len $docid $oid $eidx_key\n", $hdr or + print $w "-X $len $docid $eidx_key\n", $hdr or die "failed to write shard: $!"; } else { - $self->remove_xref3($docid, $oid, $eidx_key, $eml); + $self->remove_eidx_info($docid, $eidx_key, $eml); } } @@ -197,11 +204,22 @@ sub shard_close { } sub shard_remove { - my ($self, $oid, $num) = @_; - if (my $w = $self->{w}) { # triggers remove_by_oid in a shard child - print $w "D $oid $num\n" or die "failed to write 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_oid($oid, $num); + $self->remove_by_docid($num); + } +} + +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); } }