[PATCH] work-in-progress: crit-bit map

[PATCH] work-in-progress: crit-bit map

[Eric Wong]

Maybe I don't want to depend on SQLite?
---
 lib/PublicInbox/CBM.pm | 339 +++++++++++++++++++++++++++++++++++++++++++++++++
 t/cbm.t                |  25 ++++
 2 files changed, 364 insertions(+)
 create mode 100644 lib/PublicInbox/CBM.pm
 create mode 100644 t/cbm.t

diff --git a/lib/PublicInbox/CBM.pm b/lib/PublicInbox/CBM.pm
new file mode 100644
index 0000000..7ab583f
--- /dev/null
+++ b/lib/PublicInbox/CBM.pm
@@ -0,0 +1,339 @@
+# CC0 (Public domain): http://creativecommons.org/publicdomain/zero/1.0/
+# Note: the rest of public-inbox is AGPLv3 <meta@public-inbox.org>
+# Filesystem-based critbit implementation in pure Perl based on
+# agl's take on djb's public domain code: git://github.com/agl/critbit.git
+# Provides bidirectional mapping between MIDs and stable article nums
+# See __END__ for layout details
+package PublicInbox::CBM;
+use fields qw(fd fh);
+use strict;
+use warnings;
+use Fcntl qw(:flock SEEK_SET SEEK_END SEEK_CUR O_RDWR O_RDONLY O_CREAT O_EXCL);
+BEGIN { require 'syscall.ph' };
+use constant MY_pread => (eval { &SYS_pread64 } || eval { &SYS_pread });
+use constant MY_pwrite => (eval { &SYS_pwrite64 } || eval { &SYS_pwrite });
+use Carp qw/confess/;
+
+use constant {
+	PTR_SIZE => 8,
+	CBM_PTR => 'Q<', # 5.10 feature
+	CBM_NUM => 'Q<',
+	ALIGN_SIZE => 16,
+};
+
+use constant {
+	ALIGN_MASK => ~(ALIGN_SIZE - 1),
+	offsetof_hdr_msgnum => PTR_SIZE * 4,
+	offsetof_hdr_msgid_root => PTR_SIZE * 5,
+	offsetof_hdr_msgnum_root => PTR_SIZE * 6,
+	offsetof_hdr_free_root => PTR_SIZE * 7,
+	CBM_NULL => pack(CBM_PTR, 0),
+	offsetof_rec_mid_capa => 44,
+	offsetof_rec_num => 48,
+	sizeof_msgnum => 8,
+	sizeof_record => 64, # + msgid flex array
+};
+
+sub new {
+	my ($class, $path, $writable) = @_;
+	my $m;
+
+	if (!defined($writable) || $writable == 0) { $m = O_RDONLY; }
+	elsif ($writable == 2) { $m = O_RDWR }
+	elsif ($writable == 1) { $m = O_RDWR | O_CREAT | O_EXCL }
+	else { die "writable: `$writable' not understood\n" }
+	my $self = fields::new($class);
+
+	sysopen $self->{fh}, $path, $m or die "open $path failed: $!\n";
+	$self->{fd} = fileno($self->{fh}) + 0;
+	if ($writable == 1) {
+		my $p = sysseek($self->{fh}, 0, SEEK_END) or die "seek: $!\n";
+		# initialize the header on creation
+		my $z = "\0";
+		xpwrite($self->{fd}, $z, 63) if ($p == 0);
+	}
+	$self;
+}
+
+# msgid must be clean without '<>'
+# Returns the new msgnum if inserted, undef if it exists
+sub cbm_insert {
+	my ($self, $msgid) = @_;
+	lockrw($self);
+	my $msgnum = eval { do_insert($self, $msgid) };
+	my $err = $@;
+	unlock($self);
+	die $err if $err;
+	$msgnum;
+}
+
+sub align { ($_[0] + ALIGN_SIZE - 1) & ALIGN_MASK }
+
+sub alloc_node {
+	my ($self) = @_;
+	my $end = sysseek($self->{fh}, 0, SEEK_END) or die "seek: $!\n";
+	align($end);
+}
+
+# returns a new msgnum
+sub do_insert {
+	my ($self, $ubytes) = @_; # ubytes is a Message-ID
+	my $fd = $self->{fd};
+	my $root = xpread($fd, PTR_SIZE, offsetof_hdr_msgid_root);
+	my $ulen = bytes::length($ubytes);
+
+	# empty tree, easy
+	if ($root eq CBM_NULL) {
+		my $n = next_msgnum($fd);
+		my $newnode = alloc_node($self);
+		my $buf = pack('L<Q<A*', $ulen, $n, $ubytes);
+		xpwrite($fd, $buf, $newnode + offsetof_rec_mid_capa);
+
+		# t->root = x;
+		xpwrite($fd, pack(CBM_PTR, $newnode), offsetof_hdr_msgid_root);
+		return $n;
+	}
+
+	my @ubytes = unpack('C*', $ubytes);
+	my $p = $root = unpack(CBM_PTR, $root);
+	while (1 & $p) {
+		my ($q0, $q1, $byte, $otherbits) = mid_node($fd, $p - 1);
+		my $c = ($byte < $ulen) ? $ubytes[$byte] : 0;
+		$p = ((1 + ($otherbits | $c)) >> 8) ? $q1 : $q0;
+	}
+
+	my $newotherbits;
+	my ($nearnum, $nearmid) = node_unpack($fd, $p);
+	my @p = unpack('C*', $nearmid);
+	my $pnew;
+	my $newbyte = 0;
+	while ($newbyte < $ulen) {
+		$pnew = $p[$newbyte];
+		my $unew = $ubytes[$newbyte];
+		if ($pnew != $unew) {
+			$newotherbits = $pnew ^ $unew;
+			goto different_byte_found;
+		}
+		++$newbyte;
+	}
+	$pnew = $p[$newbyte];
+	if ($pnew != 0) {
+		$newotherbits = $pnew;
+		goto different_byte_found;
+	}
+	return undef; # duplicate
+
+different_byte_found:
+	$newotherbits |= $newotherbits >> 1;
+	$newotherbits |= $newotherbits >> 2;
+	$newotherbits |= $newotherbits >> 4;
+	$newotherbits = ($newotherbits & ~($newotherbits >> 1)) ^ 255;
+	my $c = $pnew;
+	my $newdirection = (1 + ($newotherbits | $c)) >> 8;
+	my $wherep = offsetof_hdr_msgid_root;
+	$p = $root;
+	while ($p & 1) {
+		my $q = $p - 1;
+		my ($q0, $q1, $byte, $otherbits) = mid_node($fd, $q);
+		last if ($byte > $newbyte);
+		last if ($byte == $newbyte && $otherbits > $newotherbits);
+		my $c = $byte < $ulen ? $ubytes[$byte] : 0;
+		my $direction = (1 + ($otherbits | $c)) >> 8;
+		$wherep = $direction ? $q : $q + PTR_SIZE;
+		$p = unpack(CBM_PTR, xpread($fd, PTR_SIZE, $wherep));
+	}
+
+	my $n = next_msgnum($fd);
+
+	my $newnode = alloc_node($self);
+
+	my @child = $newdirection ? ($newnode, $p) : ($p, $newnode);
+	my $buf = pack('Q<Q<Q<Q<L<L<CCSL<Q<A*',
+			$child[0], $child[1], 0, 0,
+			$newbyte, 0,
+			$newotherbits, 0, 0,
+			$ulen, $n, $ubytes);
+
+	xpwrite($fd, $buf, $newnode);
+	xpwrite($fd, pack(CBM_PTR, $newnode + 1), $wherep);
+
+	$n;
+}
+
+sub msgnum_lookup {
+	my ($fd, $umsgid) = @_;
+	my $root = xpread($fd, PTR_SIZE, offsetof_hdr_msgid_root);
+	return undef if $root eq CBM_NULL;
+	my $p = unpack(CBM_PTR, $root);
+	my @ubytes = unpack('C*', $umsgid);
+	my $ulen = scalar @ubytes;
+	while (1 & $p) {
+		my ($q0, $q1, $byte, $otherbits) = mid_node($fd, $p - 1);
+		my $c = ($byte < $ulen) ? $ubytes[$byte] : 0;
+		$p = ((1 + ($otherbits | $c)) >> 8) ? $q1 : $q0;
+	}
+	my @x;
+	my ($msgnum, $msgid) = @x = node_unpack($fd, $p);
+	($msgid eq $umsgid) ? $msgnum : undef;
+}
+
+sub msgid_lookup {
+	my ($fd, $num) = @_;
+	my $root = xpread($fd, PTR_SIZE, offsetof_hdr_msgid_root);
+	return undef if $root eq CBM_NULL;
+	my $p = unpack(CBM_PTR, $root);
+	my $ubytes = pack(CBM_PTR, $num + 0);
+	my @ubytes = unpack('C*', $ubytes);
+	my $ulen = scalar @ubytes;
+	while (1 & $p) {
+		my ($q0, $q1, $byte, $otherbits) = num_node($fd, $p - 1);
+		my $c = ($byte < $ulen) ? $ubytes[$byte] : 0;
+		$p = ((1 + ($otherbits | $c)) >> 8) ? $q1 : $q0;
+	}
+	my ($msgnum, $msgid) = node_unpack($fd, $p);
+	($msgnum == $num) ? $msgid : undef;
+}
+
+sub mid_node {
+	my ($fd, $q) = @_;
+	my @u = unpack('Q<Q<Q<Q<L<L<C', xpread($fd, 41, $q));
+	($u[0], $u[1], $u[4], $u[6]);
+}
+
+sub num_node {
+	my ($fd, $q) = @_;
+	my @u = unpack('Q<Q<L<L<CC', xpread($fd, 26, $q + 16));
+	($u[0], $u[1], $u[3], $u[5]);
+}
+
+sub node_unpack {
+	my ($fd, $q) = @_;
+	# my $q = unpack(CBM_PTR, $p);
+	my $capa = xpread($fd, 4, $q + offsetof_rec_mid_capa);
+	$capa = unpack('L<', $capa);
+	my $buf = xpread($fd, sizeof_msgnum + $capa, $q + offsetof_rec_num);
+	unpack("Q<A$capa", $buf); # (uin64_t msgnum, uint8_t msgid[])
+};
+
+# lookup the clean Message-ID given a message number
+sub cbm_msgid {
+	my ($self, $msgnum) = @_;
+	lockro($self);
+	my $msgid = msgid_lookup($self->{fd}, $msgnum);
+	unlock($self);
+	$msgid;
+}
+
+# returns the message number given a clean Message-ID
+sub cbm_msgnum {
+	my ($self, $msgid) = @_;
+	lockro($self);
+	my $msgnum = eval { msgnum_lookup($self->{fd}, $msgid) };
+	my $err = $@;
+	unlock($self);
+	die $err if $err;
+	$msgnum;
+}
+
+sub cbm_delete {
+	my ($self, $msgid) = @_; # TODO
+}
+
+sub cbm_prefixed {
+	my ($self, $msgid_pfx, $max) = @_;
+}
+
+sub next_msgnum {
+	my ($fd) = @_;
+	my $buf = xpread($fd, sizeof_msgnum, offsetof_hdr_msgnum);
+	my $num = unpack(CBM_NUM, $buf) + 1;
+	xpwrite($fd, pack(CBM_NUM, $num), offsetof_hdr_msgnum);
+	$num;
+}
+
+sub xpread {
+	# my ($fd, $count, $offset) = @_;
+	my $count = $_[1] + 0;
+	my $buf = "\0" x $count;
+	# pread(fd, buf, count, offset)
+	my $nr = syscall(MY_pread, $_[0], $buf, $count, $_[2]);
+	return $buf if $nr == $count;
+	confess "pread failed: $!\n" if $nr < 0;
+	confess "pread partial (got:$nr < expected:$count)";
+}
+
+sub xpwrite {
+	# my ($fd, $buf, $offset) = @_;
+	use bytes;
+	my $count = bytes::length($_[1]);
+	# pwrite(fd, buf, count, offset)
+	my $nr = syscall(MY_pwrite, $_[0], $_[1], $count, $_[2]);
+	return $nr if ($count == $nr);
+
+	confess "pwrite failed: $!\n" if $nr < 0;
+	confess "pwrite partial (got:$nr < expected:$count)";
+}
+
+sub lockrw { flock($_[0]->{fh}, LOCK_EX) or die "flock(LOCK_EX): $!\n" }
+sub lockro { flock($_[0]->{fh}, LOCK_SH) or die "flock(LOCK_SH): $!\n" }
+sub unlock { flock($_[0]->{fh}, LOCK_UN) or die "flock(LOCK_UN): $!\n" }
+
+1;
+__END__
+This is superior to existing on-disk databases (which I know about)
+as it allows prefix matching of Message-IDs.  Prefix matching of
+Message-IDs allows us to implement partial lookups when URLs get
+truncated.
+
+Unsigned 64-bit (LE) file offsets are "pointers" in the file,
+
+/* This is what we store and care about in the database */
+struct rec {
+	uint64_t mid_child_0;
+	uint64_t mid_child_1;
+	uint64_t num_child_0;
+	uint64_t num_child_1;
+	/* ^^ 32 bytes ^^ */
+	uint32_t mid_byte;
+	uint32_t num_byte;
+	uint8_t mid_otherbits;
+	uint8_t num_otherbits;
+	uint16_t padding;
+	uint32_t id_capa; /* max: 998, ref rfc2822 2.2.1 */
+	/* ^^ 48 bytes ^^ */
+
+	/* the actual data we care about: */
+	uint64_t num;
+	uint8_t id[]; /* flex array */
+} __attribute__((packed));
+
+# Data structure documentation in C
+struct free_node {
+	uint64_t free_child_0;
+	uint64_t free_child_1;
+	uint32_t free_byte;
+	uint8_t free_otherbits;
+	uint32_t capa;
+	uint64_t head_ptr;
+};
+
+struct any {
+	union {
+		struct free_node;
+		struct rec;
+	} as;
+};
+
+struct file_header {
+	uint64_t reserved[4];
+	uint64_t msgnum;
+	struct rec *msgid_root;
+	struct rec *msgnum_root;
+	struct free_node *free_root;
+};
+
+Note: the '*' pointer notation denotes a 64-bit LE unsigned file offset,
+not a machine/process-dependent memory pointer.
+
+This document is dedicated to the Public domain (CC0):
+http://creativecommons.org/publicdomain/zero/1.0/
diff --git a/t/cbm.t b/t/cbm.t
new file mode 100644
index 0000000..3425920
--- /dev/null
+++ b/t/cbm.t
@@ -0,0 +1,25 @@
+# CC0 (Public domain): http://creativecommons.org/publicdomain/zero/1.0/
+# Note: the rest of public-inbox is AGPLv3 <meta@public-inbox.org>
+use strict;
+use warnings;
+use Test::More;
+use File::Temp qw/tempdir/;
+
+use_ok 'PublicInbox::CBM';
+my $tmpdir = tempdir(CLEANUP => 1);
+my $f = "$tmpdir/cbm";
+my $db = PublicInbox::CBM->new($f, 1);
+
+my $n = $db->cbm_insert("hello-world");
+is(1, $n, "first message inserted is 1");
+my $c = $db->cbm_insert("ablahello-world!");
+is(2, $c, "second message inserted is 2");
+foreach my $i (1..200) {
+	is($i + 2, $db->cbm_insert("msg".$i), "insert is sequential ($i)");
+}
+#print STDERR $db->cbm_msgid(2), "\n";
+is(1, $db->cbm_msgnum("hello-world"), "msgid lookup works");
+# system "cat $f | hexdump -C >&2";
+
+done_testing();
+1;
-- 
EW