X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FSearchIdxShard.pm;h=45240e0793a8867e272005d386712e81004af803;hb=af0b0fb7a454470a32c452119d0392e0dedb3fe1;hp=e194b7e0aa1f0ab2e3e464e6fbc163ea36655824;hpb=5a1ee8a928382bc3fac9b8faff67b0c24396ff9b;p=public-inbox.git

diff --git a/lib/PublicInbox/SearchIdxShard.pm b/lib/PublicInbox/SearchIdxShard.pm
index e194b7e0..45240e07 100644
--- a/lib/PublicInbox/SearchIdxShard.pm
+++ b/lib/PublicInbox/SearchIdxShard.pm
@@ -1,4 +1,4 @@
-# Copyright (C) 2018-2020 all contributors <meta@public-inbox.org>
+# Copyright (C) 2018-2021 all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
 # Internal interface for a single Xapian shard in V2 inboxes.
@@ -10,6 +10,7 @@ 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) = @_; # v2w may be ExtSearchIdx
@@ -29,9 +30,14 @@ 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) {
+		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: $!";
 
@@ -44,6 +50,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: $!";
@@ -51,6 +58,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);
@@ -62,34 +70,41 @@ sub shard_worker_loop ($$$$$) {
 	$0 = "shard[$shard]";
 	$self->begin_txn_lazy;
 	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") {
+		} elsif ($line eq 'close') {
 			$self->idx_release;
-		} elsif ($line eq "barrier\n") {
+		} 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,}) ([0-9]+)\n\z/s) {
-			$self->remove_by_oid($1, $2 + 0);
+		} elsif ($line =~ /\AD ([0-9]+)\z/s) {
+			$self->remove_by_docid($1 + 0);
 		} elsif ($line =~ s/\A\+X //) {
-			my ($len, $docid, $oid, $eidx_key) =
-							split(/ /, $line, 4);
-			$self->add_eidx_info($docid, $oid, $eidx_key,
-							eml($r, $len));
+			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, $oid, $eidx_key) =
-							split(/ /, $line, 4);
-			$self->remove_eidx_info($docid, $oid, $eidx_key,
+			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 $eidx_key;
 			if ($line =~ s/\AX=(.+)\0//) {
 				$eidx_key = $1;
-				$v2w->{current_info} =~ s/\0/\\0/;
+				$v2w->{current_info} =~ s/\0/\\0 /;
 			}
 			# n.b. $mid may contain spaces(!)
 			my ($len, $bytes, $num, $oid, $ds, $ts, $tid, $mid)
@@ -112,16 +127,13 @@ sub shard_worker_loop ($$$$$) {
 }
 
 sub index_raw {
-	my ($self, $msgref, $eml, $smsg, $ibx) = @_;
+	my ($self, $msgref, $eml, $smsg, $eidx_key) = @_;
 	if (my $w = $self->{w}) {
-		if ($ibx) {
-			print $w 'X=', $ibx->eidx_key, "\0" or die
-				"failed to write shard: $!\n";
-		}
+		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 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 {
@@ -131,34 +143,32 @@ sub index_raw {
 			$eml = PublicInbox::Eml->new($msgref);
 		}
 		$self->begin_txn_lazy;
-		$smsg->{eidx_key} = $ibx->eidx_key if $ibx;
+		$smsg->{eidx_key} = $eidx_key if defined $eidx_key;
 		$self->add_message($eml, $smsg);
 	}
 }
 
 sub shard_add_eidx_info {
-	my ($self, $docid, $oid, $xibx, $eml) = @_;
-	my $eidx_key = $xibx->eidx_key;
+	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 $oid $eidx_key\n", $hdr or
+		print $w "+X $len $docid $eidx_key\n", $hdr or
 			die "failed to write shard: $!";
 	} else {
-		$self->add_eidx_info($docid, $oid, $eidx_key, $eml);
+		$self->add_eidx_info($docid, $eidx_key, $eml);
 	}
 }
 
 sub shard_remove_eidx_info {
-	my ($self, $docid, $oid, $xibx, $eml) = @_;
-	my $eidx_key = $xibx->eidx_key;
+	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_eidx_info($docid, $oid, $eidx_key, $eml);
+		$self->remove_eidx_info($docid, $eidx_key, $eml);
 	}
 }
 
@@ -199,11 +209,49 @@ 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_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->remove_by_oid($oid, $num);
+		$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);
 	}
 }