lib/PublicInbox/SaPlugin/ListMirror.pm
lib/PublicInbox/Search.pm
lib/PublicInbox/SearchIdx.pm
+lib/PublicInbox/SearchIdxPart.pm
+lib/PublicInbox/SearchIdxThread.pm
lib/PublicInbox/SearchMsg.pm
lib/PublicInbox/SearchThread.pm
lib/PublicInbox/SearchView.pm
$self->{bytes_added} += $n;
print $w "blob\nmark :$blob\ndata ", $n, "\n" or wfail;
print $w $str, "\n" or wfail;
- $str = undef;
# v2: we need this for Xapian
if ($self->{want_object_id}) {
chomp($self->{last_object_id} = $self->get_mark(":$blob"));
- $self->{last_object_size} = $n;
+ $self->{last_object} = [ $n, \$str ];
}
-
my $ref = $self->{ref};
my $commit = $self->{mark}++;
my $parent = $tip =~ /\A:/ ? $tip : undef;
if ($self->{version} == 1) {
"$self->{mainrepo}/public-inbox/xapian" . SCHEMA_VERSION;
} else {
- "$self->{mainrepo}/xap" . SCHEMA_VERSION;
+ my $dir = "$self->{mainrepo}/xap" . SCHEMA_VERSION;
+ my $part = $self->{partition};
+ defined $part or die "partition not given";
+ $dir .= "/$part";
}
}
}
sub new {
- my ($class, $ibx, $creat) = @_;
+ my ($class, $ibx, $creat, $part) = @_;
my $mainrepo = $ibx; # for "public-inbox-index" w/o entry in config
my $git_dir = $mainrepo;
my ($altid, $git);
if ($version == 1) {
$self->{lock_path} = "$mainrepo/ssoma.lock";
} elsif ($version == 2) {
- $self->{lock_path} = "$mainrepo/inbox.lock";
+ defined $part or die "partition is required for v2\n";
+ # partition is a number or "all"
+ $self->{partition} = $part;
+ $self->{lock_path} = undef;
$self->{msgmap_path} = "$mainrepo/msgmap.sqlite3";
} else {
die "unsupported inbox version=$version\n";
sub _lock_acquire {
my ($self) = @_;
croak 'already locked' if $self->{lockfh};
- sysopen(my $lockfh, $self->{lock_path}, O_WRONLY|O_CREAT) or
- die "failed to open lock $self->{lock_path}: $!\n";
+ my $lock_path = $self->{lock_path} or return;
+ sysopen(my $lockfh, $lock_path, O_WRONLY|O_CREAT) or
+ die "failed to open lock $lock_path: $!\n";
flock($lockfh, LOCK_EX) or die "lock failed: $!\n";
$self->{lockfh} = $lockfh;
}
sub _lock_release {
my ($self) = @_;
+ return unless $self->{lock_path};
my $lockfh = delete $self->{lockfh} or croak 'not locked';
flock($lockfh, LOCK_UN) or die "unlock failed: $!\n";
close $lockfh or die "close failed: $!\n";
$doc->add_value($col, $num);
}
-sub add_values ($$$) {
- my ($smsg, $bytes, $num) = @_;
+sub add_values ($$$$) {
+ my ($smsg, $bytes, $num, $lines) = @_;
my $ts = $smsg->ts;
my $doc = $smsg->{doc};
defined($bytes) and add_val($doc, &PublicInbox::Search::BYTES, $bytes);
- add_val($doc, &PublicInbox::Search::LINES,
- $smsg->{mime}->body_raw =~ tr!\n!\n!);
+ add_val($doc, &PublicInbox::Search::LINES, $lines);
my $yyyymmdd = strftime('%Y%m%d', gmtime($ts));
add_val($doc, PublicInbox::Search::YYYYMMDD, $yyyymmdd);
my ($doc_id, $old_tid);
my $mid = mid_clean(mid_mime($mime));
+ my $threader = $self->{threader};
eval {
die 'Message-ID too long' if length($mid) > MAX_MID_SIZE;
# convert a ghost to a regular message
# it will also clobber any existing regular message
$doc_id = $smsg->{doc_id};
- $old_tid = $smsg->thread_id;
+ $old_tid = $smsg->thread_id unless $threader;
}
$smsg = PublicInbox::SearchMsg->new($mime);
my $doc = $smsg->{doc};
$doc->add_term('XMID' . $mid);
my $subj = $smsg->subject;
+ my $xpath;
if ($subj ne '') {
- my $path = $self->subject_path($subj);
- $doc->add_term('XPATH' . id_compress($path));
+ $xpath = $self->subject_path($subj);
+ $xpath = id_compress($xpath);
+ $doc->add_term('XPATH' . $xpath);
}
- add_values($smsg, $bytes, $num);
+ my $lines = $mime->body_raw =~ tr!\n!\n!;
+ add_values($smsg, $bytes, $num, $lines);
my $tg = $self->term_generator;
index_body($tg, \@orig, $doc) if @orig;
});
- link_message($self, $smsg, $old_tid);
+ # populates smsg->references for smsg->to_doc_data
+ my $refs = parse_references($smsg);
+ my $data = $smsg->to_doc_data($blob);
+ if ($threader) {
+ $threader->thread_msg($mid, $smsg->ts, $xpath, $data);
+ } else {
+ link_message($self, $smsg, $refs, $old_tid);
+ }
$tg->index_text($mid, 1, 'XM');
- $doc->set_data($smsg->to_doc_data($blob));
+ $doc->set_data($data);
if (my $altid = $self->{-altid}) {
foreach my $alt (@$altid) {
$last_thread_id;
}
-sub link_message {
- my ($self, $smsg, $old_tid) = @_;
+sub parse_references ($) {
+ my ($smsg) = @_;
my $doc = $smsg->{doc};
my $mid = $smsg->mid;
my $mime = $smsg->{mime};
my @refs = (($hdr->header_raw('References') || '') =~ /<([^>]+)>/g);
push(@refs, (($hdr->header_raw('In-Reply-To') || '') =~ /<([^>]+)>/g));
- my $tid;
if (@refs) {
my %uniq = ($mid => 1);
my @orig_refs = @refs;
push @refs, $ref;
}
}
+ $smsg->{references} = '<'.join('> <', @refs).'>' if @refs;
+ \@refs
+}
- if (@refs) {
- $smsg->{references} = '<'.join('> <', @refs).'>';
+sub link_message {
+ my ($self, $smsg, $refs, $old_tid) = @_;
+ my $tid;
+
+ if (@$refs) {
# first ref *should* be the thread root,
# but we can never trust clients to do the right thing
- my $ref = shift @refs;
+ my $ref = shift @$refs;
$tid = $self->_resolve_mid_to_tid($ref);
$self->merge_threads($tid, $old_tid) if defined $old_tid;
# the rest of the refs should point to this tid:
- foreach $ref (@refs) {
+ foreach $ref (@$refs) {
my $ptid = $self->_resolve_mid_to_tid($ref);
merge_threads($self, $tid, $ptid);
}
} else {
$tid = defined $old_tid ? $old_tid : $self->next_thread_id;
}
- $doc->add_term('G' . $tid);
+ $smsg->{doc}->add_term('G' . $tid);
}
sub index_blob {
$_[0]->{lockfh} = undef;
}
+# remote_* subs are only used by SearchIdxPart and SearchIdxThread:
+sub remote_commit {
+ my ($self) = @_;
+ print { $self->{w} } "commit\n" or die "failed to write commit: $!";
+}
+
+sub remote_close {
+ my ($self) = @_;
+ my $pid = delete $self->{pid} or die "no process to wait on\n";
+ my $w = delete $self->{w} or die "no pipe to write to\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)." exited with: $?";
+}
+
1;
--- /dev/null
+# Copyright (C) 2018 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+package PublicInbox::SearchIdxPart;
+use strict;
+use warnings;
+use base qw(PublicInbox::SearchIdx);
+
+sub new {
+ my ($class, $v2writable, $part, $threader) = @_;
+ my $self = $class->SUPER::new($v2writable->{-inbox}, 1, $part);
+ $self->{threader} = $threader;
+ my ($r, $w);
+ pipe($r, $w) or die "pipe failed: $!\n";
+ my $pid = fork;
+ defined $pid or die "fork failed: $!\n";
+ if ($pid == 0) {
+ foreach my $other (@{$v2writable->{idx_parts}}) {
+ my $other_w = $other->{w} or next;
+ close $other_w or die "close other failed: $!\n";
+ }
+ $v2writable = undef;
+ close $w;
+ eval { partition_worker_loop($self, $r) };
+ die "worker $part died: $@\n" if $@;
+ die "unexpected MM $self->{mm}" if $self->{mm};
+ exit;
+ }
+ $self->{pid} = $pid;
+ $self->{w} = $w;
+ close $r;
+ $self;
+}
+
+sub partition_worker_loop ($$) {
+ my ($self, $r) = @_;
+ my $xdb = $self->_xdb_acquire;
+ $xdb->begin_transaction;
+ my $txn = 1;
+ while (my $line = $r->getline) {
+ if ($line eq "commit\n") {
+ $xdb->commit_transaction if $txn;
+ $txn = undef;
+ } elsif ($line eq "close\n") {
+ $self->_xdb_release;
+ $xdb = $txn = undef;
+ } else {
+ my ($len, $artnum, $object_id) = split(/ /, $line);
+ $xdb ||= $self->_xdb_acquire;
+ if (!$txn) {
+ $xdb->begin_transaction;
+ $txn = 1;
+ }
+ my $n = read($r, my $msg, $len) or die "read: $!\n";
+ $n == $len or die "short read: $n != $len\n";
+ my $mime = PublicInbox::MIME->new(\$msg);
+ $self->index_blob($mime, $len, $artnum, $object_id);
+ }
+ }
+ warn "$$ still in transaction\n" if $txn;
+ warn "$$ xdb active\n" if $xdb;
+}
+
+# called by V2Writable
+sub index_raw {
+ my ($self, $len, $msgref, $artnum, $object_id) = @_;
+ print { $self->{w} } "$len $artnum $object_id\n", $$msgref or die
+ "failed to write partition $!\n";
+}
+
+1;
--- /dev/null
+# Copyright (C) 2018 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+package PublicInbox::SearchIdxThread;
+use strict;
+use warnings;
+use base qw(PublicInbox::SearchIdx);
+use Storable qw(freeze thaw);
+
+sub new {
+ my ($class, $v2ibx) = @_;
+ my $self = $class->SUPER::new($v2ibx, 1, 'all');
+ # create the DB:
+ $self->_xdb_acquire;
+ $self->_xdb_release;
+
+ my ($r, $w);
+ pipe($r, $w) or die "pipe failed: $!\n";
+ binmode $r, ':raw';
+ binmode $w, ':raw';
+ my $pid = fork;
+ defined $pid or die "fork failed: $!\n";
+ if ($pid == 0) {
+ close $w;
+ eval { thread_worker_loop($self, $r) };
+ die "thread worker died: $@\n" if $@;
+ exit;
+ }
+ $self->{w} = $w;
+ $self->{pid} = $pid;
+ close $r;
+
+ $w->autoflush(1);
+
+ # lock on only exists in parent, not in worker
+ my $l = $self->{lock_path} = $self->xdir . '/thread.lock';
+ open my $fh, '>>', $l or die "failed to create $l: $!\n";
+ $self;
+}
+
+sub thread_worker_loop {
+ my ($self, $r) = @_;
+ my $msg;
+ my $xdb = $self->_xdb_acquire;
+ $xdb->begin_transaction;
+ my $txn = 1;
+ while (my $line = $r->getline) {
+ if ($line eq "commit\n") {
+ $xdb->commit_transaction if $txn;
+ $txn = undef;
+ } elsif ($line eq "close\n") {
+ $self->_xdb_release;
+ $xdb = $txn = undef;
+ } else {
+ read($r, $msg, $line) or die "read failed: $!\n";
+ $msg = thaw($msg); # should raise on error
+ defined $msg or die "failed to thaw buffer\n";
+ if (!$txn) {
+ $xdb->begin_transaction;
+ $txn = 1;
+ }
+ eval { $self->thread_msg_real(@$msg) };
+ warn "failed to index message <$msg->[0]>: $@\n" if $@;
+ }
+ }
+}
+
+# called by a partition worker
+sub thread_msg {
+ my ($self, $mid, $ts, $xpath, $doc_data) = @_;
+ my $w = $self->{w};
+ my $err;
+ my $str = freeze([ $mid, $ts, $xpath, $doc_data ]);
+ my $len = length($str) . "\n";
+
+ # multiple processes write to the same pipe, so use flock
+ $self->_lock_acquire;
+ print $w $len, $str or $err = $!;
+ $self->_lock_release;
+
+ die "print failed: $err\n" if $err;
+}
+
+sub thread_msg_real {
+ my ($self, $mid, $ts, $xpath, $doc_data) = @_;
+ my $smsg = $self->lookup_message($mid);
+ my ($old_tid, $doc_id);
+ if ($smsg) {
+ # convert a ghost to a regular message
+ # it will also clobber any existing regular message
+ $doc_id = $smsg->{doc_id};
+ $old_tid = $smsg->thread_id;
+ } else {
+ $smsg = PublicInbox::SearchMsg->new(undef);
+ $smsg->{mid} = $mid;
+ }
+ my $doc = $smsg->{doc};
+ $doc->add_term('XPATH' . $xpath) if defined $xpath;
+ $doc->add_term('XMID' . $mid);
+ $doc->set_data($doc_data);
+ $smsg->{ts} = $ts;
+ my @refs = ($smsg->references =~ /<([^>]+)>/g);
+ $self->link_message($smsg, \@refs, $old_tid);
+ my $db = $self->{xdb};
+ if (defined $doc_id) {
+ $db->replace_document($doc_id, $doc);
+ } else {
+ $doc_id = $db->add_document($doc);
+ }
+}
+
+1;
Search::Xapian::sortable_unserialise($doc->get_value($col));
}
-sub load_expand {
- my ($self) = @_;
- my $doc = $self->{doc};
- my $data = $doc->get_data or return;
- $self->{ts} = get_val($doc, &PublicInbox::Search::TS);
- utf8::decode($data);
- my ($subj, $from, $refs, $to, $cc, $blob) = split(/\n/, $data);
+sub load_from_data ($$) {
+ my ($self) = $_[0]; # data = $_[1]
+ my ($subj, $from, $refs, $to, $cc, $blob) = split(/\n/, $_[1]);
$self->{subject} = $subj;
$self->{from} = $from;
$self->{references} = $refs;
$self->{to} = $to;
$self->{cc} = $cc;
$self->{blob} = $blob;
+}
+
+sub load_expand {
+ my ($self) = @_;
+ my $doc = $self->{doc};
+ my $data = $doc->get_data or return;
+ $self->{ts} = get_val($doc, &PublicInbox::Search::TS);
+ utf8::decode($data);
+ load_from_data($self, $data);
$self;
}
my $data = $doc->get_data or return;
my $ts = get_val($doc, &PublicInbox::Search::TS);
utf8::decode($data);
- my ($subj, $from, $refs, $to, $cc, $blob) = split(/\n/, $data);
- bless {
- doc => $doc,
- subject => $subj,
- ts => $ts,
- from => $from,
- references => $refs,
- to => $to,
- cc => $cc,
- blob => $blob,
- }, $class;
+ my $self = bless { doc => $doc, ts => $ts }, $class;
+ load_from_data($self, $data);
+ $self
}
# :bytes and :lines metadata in RFC 3977
use strict;
use warnings;
use Fcntl qw(:flock :DEFAULT);
-use PublicInbox::SearchIdx;
+use PublicInbox::SearchIdxPart;
+use PublicInbox::SearchIdxThread;
use PublicInbox::MIME;
use PublicInbox::Git;
use PublicInbox::Import;
im => undef, # PublicInbox::Import
xap_rw => undef, # PublicInbox::V2SearchIdx
xap_ro => undef,
-
+ partitions => 4,
+ transact_bytes => 0,
# limit each repo to 1GB or so
rotate_bytes => int((1024 * 1024 * 1024) / $PACKING_FACTOR),
};
my $cmt = $im->add($mime, $check_cb) or return;
$cmt = $im->get_mark($cmt);
my $oid = $im->{last_object_id};
- my $size = $im->{last_object_size};
-
- my $idx = $self->search_idx;
- $idx->index_both($mime, $size, $oid);
- $idx->{xdb}->set_metadata('last_commit', $cmt);
- my $n = $self->{transact_bytes} += $size;
- if ($n > PublicInbox::SearchIdx::BATCH_BYTES) {
+ my ($len, $msgref) = @{$im->{last_object}};
+
+ my $nparts = $self->{partitions};
+ my $part = hex(substr($oid, 0, 8)) % $nparts;
+ my $idx = $self->idx_part($part);
+ my $all = $self->{all};
+ my $num = $all->index_mm($mime);
+ $idx->index_raw($len, $msgref, $num, $oid);
+ my $n = $self->{transact_bytes} += $len;
+ if ($n > (PublicInbox::SearchIdx::BATCH_BYTES * $nparts)) {
$self->checkpoint;
}
$mime;
}
-sub search_idx {
- my ($self) = @_;
- $self->{idx} ||= eval {
- my $idx = PublicInbox::SearchIdx->new($self->{-inbox}, 1);
- my $mm = $idx->_msgmap_init;
- $idx->_xdb_acquire->begin_transaction;
- $self->{transact_bytes} = 0;
- $mm->{dbh}->begin_work;
- $idx
- };
+sub idx_part {
+ my ($self, $part) = @_;
+ my $idx = $self->{idx_parts};
+ return $idx->[$part] if $idx; # fast path
+
+ # first time initialization:
+ my $all = $self->{all} =
+ PublicInbox::SearchIdxThread->new($self->{-inbox});
+
+ # need to create all parts before initializing msgmap FD
+ my $max = $self->{partitions} - 1;
+ $idx = $self->{idx_parts} = [];
+ for my $i (0..$max) {
+ push @$idx, PublicInbox::SearchIdxPart->new($self, $i, $all);
+ }
+ $all->_msgmap_init->{dbh}->begin_work;
+ $idx->[$part];
}
sub remove {
my ($self) = @_;
my $im = $self->{im};
$im->done if $im; # PublicInbox::Import::done
- $self->searchidx_checkpoint;
+ $self->searchidx_checkpoint(0);
}
sub checkpoint {
my ($self) = @_;
my $im = $self->{im};
$im->checkpoint if $im; # PublicInbox::Import::checkpoint
- $self->searchidx_checkpoint;
+ $self->searchidx_checkpoint(1);
}
sub searchidx_checkpoint {
- my ($self) = @_;
- my $idx = delete $self->{idx} or return;
+ my ($self, $more) = @_;
+
+ # order matters, we can only close {all} after all partitions
+ # are done because the partitions also write to {all}
+
+ my $parts = $self->{idx_parts};
+ foreach my $idx (@$parts) {
+ $idx->remote_commit;
+ $idx->remote_close unless $more;
+ }
- $idx->{mm}->{dbh}->commit;
- $idx->{xdb}->commit_transaction;
- $idx->_xdb_release;
+ if (my $all = $self->{all}) {
+ $all->{mm}->{dbh}->commit;
+ if ($more) {
+ $all->{mm}->{dbh}->begin_work;
+ }
+ $all->remote_commit;
+ $all->remote_close unless $more;
+ }
+ $self->{transact_bytes} = 0;
}
sub git_init {
} else {
$self->{im} = undef;
$im->done;
- $self->searchidx_checkpoint;
+ $self->searchidx_checkpoint(1);
$im = undef;
my $git_dir = $self->git_init(++$self->{max_git});
my $git = PublicInbox::Git->new($git_dir);