lib/PublicInbox/SharedKV.pm

   1 # Copyright (C) 2020 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3
   4 # fork()-friendly key-value store.  Will be used for making
   5 # augmenting Maildirs and mboxes less expensive, maybe.
   6 # We use flock(2) to avoid SQLite lock problems (busy timeouts, backoff)
   7 package PublicInbox::SharedKV;
   8 use strict;
   9 use v5.10.1;
  10 use parent qw(PublicInbox::Lock);
  11 use File::Temp 0.19 (); # 0.19 for ->newdir
  12 use DBI ();
  13 use PublicInbox::Spawn;
  14
  15 sub dbh {
  16         my ($self, $lock) = @_;
  17         $self->{dbh} //= do {
  18                 my $f = $self->{filename};
  19                 $lock //= $self->lock_for_scope;
  20                 my $dbh = DBI->connect("dbi:SQLite:dbname=$f", '', '', {
  21                         AutoCommit => 1,
  22                         RaiseError => 1,
  23                         PrintError => 0,
  24                         sqlite_use_immediate_transaction => 1,
  25                         # no sqlite_unicode here, this is for binary data
  26                 });
  27                 my $opt = $self->{opt} // {};
  28                 $dbh->do('PRAGMA synchronous = OFF') if !$opt->{fsync};
  29                 $dbh->do('PRAGMA cache_size = '.($opt->{cache_size} || 80000));
  30                 $dbh->do('PRAGMA journal_mode = '.
  31                                 ($opt->{journal_mode} // 'WAL'));
  32                 $dbh->do(<<'');
  33 CREATE TABLE IF NOT EXISTS kv (
  34         k VARBINARY PRIMARY KEY NOT NULL,
  35         v VARBINARY NOT NULL,
  36         UNIQUE (k)
  37 )
  38
  39                 $dbh->do('CREATE INDEX IF NOT EXISTS idx_v ON kv (v)');
  40                 $dbh;
  41         }
  42 }
  43
  44 sub new {
  45         my ($cls, $dir, $base, $opt) = @_;
  46         my $self = bless { opt => $opt }, $cls;
  47         unless (defined $dir) {
  48                 $self->{tmp} = File::Temp->newdir('kv-XXXXXX', TMPDIR => 1);
  49                 $dir = $self->{tmp}->dirname;
  50         }
  51         -d $dir or mkdir($dir) or die "mkdir($dir): $!";
  52         $base //= '';
  53         my $f = $self->{filename} = "$dir/$base.sqlite3";
  54         $self->{lock_path} = $opt->{lock_path} // "$dir/$base.flock";
  55         unless (-f $f) {
  56                 open my $fh, '+>>', $f or die "failed to open $f: $!";
  57                 PublicInbox::Spawn::nodatacow_fd(fileno($fh));
  58         }
  59         $self;
  60 }
  61
  62 sub set_maybe {
  63         my ($self, $key, $val, $lock) = @_;
  64         $lock //= $self->lock_for_scope;
  65         my $e = $self->{dbh}->prepare_cached(<<'')->execute($key, $val);
  66 INSERT OR IGNORE INTO kv (k,v) VALUES (?, ?)
  67
  68         $e == 0 ? undef : $e;
  69 }
  70
  71 # caller calls sth->fetchrow_array
  72 sub each_kv_iter {
  73         my ($self) = @_;
  74         my $sth = $self->{dbh}->prepare_cached(<<'', undef, 1);
  75 SELECT k,v FROM kv
  76
  77         $sth->execute;
  78         $sth
  79 }
  80
  81 sub delete_by_val {
  82         my ($self, $val, $lock) = @_;
  83         $lock //= $self->lock_for_scope;
  84         $self->{dbh}->prepare_cached(<<'')->execute($val) + 0;
  85 DELETE FROM kv WHERE v = ?
  86
  87 }
  88
  89 sub replace_values {
  90         my ($self, $oldval, $newval, $lock) = @_;
  91         $lock //= $self->lock_for_scope;
  92         $self->{dbh}->prepare_cached(<<'')->execute($newval, $oldval) + 0;
  93 UPDATE kv SET v = ? WHERE v = ?
  94
  95 }
  96
  97 sub set {
  98         my ($self, $key, $val) = @_;
  99         if (defined $val) {
 100                 my $e = $self->{dbh}->prepare_cached(<<'')->execute($key, $val);
 101 INSERT OR REPLACE INTO kv (k,v) VALUES (?,?)
 102
 103                 $e == 0 ? undef : $e;
 104         } else {
 105                 $self->{dbh}->prepare_cached(<<'')->execute($key);
 106 DELETE FROM kv WHERE k = ?
 107
 108         }
 109 }
 110
 111 sub get {
 112         my ($self, $key) = @_;
 113         my $sth = $self->{dbh}->prepare_cached(<<'', undef, 1);
 114 SELECT v FROM kv WHERE k = ?
 115
 116         $sth->execute($key);
 117         $sth->fetchrow_array;
 118 }
 119
 120 sub xchg {
 121         my ($self, $key, $newval, $lock) = @_;
 122         $lock //= $self->lock_for_scope;
 123         my $oldval = get($self, $key);
 124         if (defined $newval) {
 125                 set($self, $key, $newval);
 126         } else {
 127                 $self->{dbh}->prepare_cached(<<'')->execute($key);
 128 DELETE FROM kv WHERE k = ?
 129
 130         }
 131         $oldval;
 132 }
 133
 134 sub count {
 135         my ($self) = @_;
 136         my $sth = $self->{dbh}->prepare_cached(<<'');
 137 SELECT COUNT(k) FROM kv
 138
 139         $sth->execute;
 140         $sth->fetchrow_array;
 141 }
 142
 143 1;