File Coverage

blib/lib/MMapDB.pm

Criterion	Covered	Total	%
statement	420	643	65.3
branch	99	282	35.1
condition	11	65	16.9
subroutine	53	86	61.6
pod	20	20	100.0
total	603	1096	55.0

package MMapDB;

use 5.008008;
use strict;
use warnings;
no warnings qw/uninitialized/;

# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
# keep this in mind
use integer;
# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

use Fcntl qw/:seek :flock/;
use File::Spec;
use File::Map qw/map_handle protect/;
use Exporter qw/import/;
use Encode ();

{				# limit visibility of "our"/"my" variables
  our $VERSION = '0.15';
  our %EXPORT_TAGS=
    (
     mode =>[qw/DATAMODE_NORMAL DATAMODE_SIMPLE/],
     error=>[qw/E_READONLY E_TWICE E_TRANSACTION E_FULL E_DUPLICATE
		E_OPEN E_READ E_WRITE E_CLOSE E_RENAME E_TRUNCATE E_LOCK
		E_RANGE E_NOT_IMPLEMENTED/],
    );
  my %seen;
  undef @seen{map {@$_} values %EXPORT_TAGS};
  our @EXPORT_OK=keys %seen;
  $EXPORT_TAGS{all}=\@EXPORT_OK;

  require XSLoader;
  XSLoader::load('MMapDB', $VERSION);

  our @attributes;
  BEGIN {
    # define attributes and implement accessor methods
    # !! keep in sync with MMapDB.xs !!
    @attributes=(qw/filename readonly intfmt _data _intsize _stringfmt
		    _stringtbl mainidx _ididx main_index id_index
		    _nextid _idmap _tmpfh _tmpname _stringfh _stringmap
		    _strpos lockfile flags dbformat_in dbformat_out
		    _stringfmt_out stringmap_prealloc _stringmap_end
		    index_prealloc _index_end _tmpmap
		   /);
    for( my $i=0; $i<@attributes; $i++ ) {
      my $method_num=$i;
      ## no critic
      no strict 'refs';
      *{__PACKAGE__.'::'.$attributes[$method_num]}=
	sub : lvalue {$_[0]->[$method_num]};
      ## use critic
    }
  }
}

my @dbformats=qw/MMDB MMDC/;
my %dbformats=do { my $i=0; map {($_=>$i++)} @dbformats };

BEGIN {
  use constant {
    FORMATVERSION => 0,		# magic number position (in bytes)
    INTFMT        => 4,		# INTFMT byte position (in bytes)
    BASEOFFSET    => 8,
    MAINIDX       => 0,		# (in words (units of _intsize bytes))
    IDIDX         => 1,		# (in words)
    NEXTID        => 2,		# (in words)
    STRINGTBL     => 3,		# (in words)
    DATASTART     => 4,		# (in words)

    DBFMT0        => 0,		# MMDB format
    DBFMT1        => 1,		# MMDC format with utf8 support

    # iterator questions
    IT_NTH        =>0,		# reposition iterator
    IT_CUR        =>1, 		# what is the current index
    IT_NELEM      =>2,          # how many elements does it iterate over

    DATAMODE_NORMAL=>0,
    DATAMODE_SIMPLE=>1,

    E_READONLY    => \'database is read-only',
    E_TWICE       => \'can\'t insert the same ID twice',
    E_TRANSACTION => \'there is already an active transaction',
    E_FULL        => \'can\'t allocate ID',
    E_DUPLICATE   => \'data records cannot be mixed up with subindices',

    E_OPEN        => \'can\'t open file',
    E_READ        => \'can\'t read from file',
    E_WRITE       => \'can\'t write to file',
    E_CLOSE       => \'file could not be closed',
    E_RENAME      => \'can\'t rename file',
    E_SEEK        => \'can\'t move file pointer',
    E_TRUNCATE    => \'can\'t truncate file',
    E_LOCK        => \'can\'t (un)lock lockfile',
    E_RANGE       => \'attempt move iterator out of its range',
    E_NOT_IMPLEMENTED => \'function not implemented',
  };
}

#sub D {
#  use Data::Dumper;
#  local $Data::Dumper::Useqq=1;
#  warn Dumper @_;
#}

sub _putdata {
  my ($I, $pos, $fmt, @param)=@_;

  my $pstr=pack $fmt, @param;
  my $map=$I->_tmpmap;
  if( $pos+length($pstr)>length $$map ) {
    my $prea=$I->index_prealloc;
    my $need=$prea*(($pos+length($pstr)+$prea-1)/$prea);
    eval {
      my $fh=$I->_tmpfh;
      sysseek $fh, $need, SEEK_SET and
	truncate $fh, $need and
	  map_handle $$map, $fh, '+>', 0, $need;
    };
    $I->_e(E_OPEN) if $@;
  }
  substr $$map, $pos, length($pstr), $pstr;
  return length($pstr);
}

sub set_intfmt {
  my ($I, $fmt)=@_;

  $fmt='N' unless $fmt;

  my %allowed; undef @allowed{qw/L N J Q/};
  return unless exists $allowed{$fmt};

  $I->intfmt=$fmt;
  $I->_intsize=length pack($fmt, 0);

  if( $I->dbformat_in>DBFMT0 ) {
    # new format with utf8 support
    $I->_stringfmt=$I->intfmt.'/a*C x!'.$I->_intsize;
  } else {
    $I->_stringfmt=$I->intfmt.'/a* x!'.$I->_intsize;
  }

  if( $I->dbformat_out>DBFMT0 ) {
    # new format with utf8 support
    $I->_stringfmt_out=$I->intfmt.'/a*C x!'.$I->_intsize;
  } else {
    $I->_stringfmt_out=$I->intfmt.'/a* x!'.$I->_intsize;
  }

  return 1;
}

sub new {
  my ($parent, @param)=@_;
  my $I;

  if (ref $parent) {
    $I=bless [@$parent]=>ref($parent);
    for my $k (qw/_nextid _idmap _tmpfh _tmpname _stringfh _stringmap
		  _strpos main_index id_index/) {
      undef $I->$k;
    }
    if( defined $I->_data ) {
      # parameters: PARENT POS DATAMODE
      tie %{$I->main_index=+{}}, 'MMapDB::Index', $I, $I->mainidx, 0;
      tie %{$I->id_index=+{}}, 'MMapDB::IDIndex', $I, undef, 0;
    }
  } else {
    $I=bless []=>$parent;
    $I->set_intfmt('N');
    $I->flags=0;
    $I->dbformat_in=$#dbformats; # use the newest by default
    $I->dbformat_out=$#dbformats; # use the newest by default
  }
  $I->stringmap_prealloc=1024*1024*10; # 10MB
  $I->index_prealloc=1024*1024*10; # 10MB

  if( @param==1 ) {
    $I->filename=$param[0];
  } else {
    while( my ($k, $v)=splice @param, 0, 2 ) {
      $I->$k=$v if $k=$I->can($k);
    }
    $I->set_intfmt($I->intfmt) unless $I->intfmt eq 'N';
  }

  return $I;
}

sub is_valid {
  my ($I)=@_;

  return unless $I->_data;
  # the INTFMT field serves 2 ways:
  #  1) it specifies the used integer format
  #  2) it works as VALID flag. commit() write a NULL byte here
  #     to mark the old file as invalid.
  #     we must reconnect if our cached fmt does not match.
  return substr( ${$I->_data}, INTFMT, 1 ) eq $I->intfmt;
}

sub start {
  my ($I)=@_;

  $I->_e(E_TRANSACTION) if defined $I->_tmpfh;

  my $retry=5;
  RETRY: {
      return unless $retry--;
      $I->stop if (defined $I->_data and
		   substr( ${$I->_data}, INTFMT, 1 ) ne $I->intfmt);

      unless( $I->_data ) {
	my ($dummy, $fmt);
	my $fh;
	if( $I->readonly ) {
	  open $fh, '<', $I->filename or return;
	} else {
	  open $fh, '+<', $I->filename or return;
	}

	# Map the main data always read-only. If we are in writable mode
	# map only the header page again writable.
	eval {
	  map_handle $dummy, $fh, '<';
	};
	close $fh;
	return if $@;		# perhaps throw something here
	return unless length $dummy;

	# check magic number
	return unless exists $dbformats{substr($dummy, FORMATVERSION, 4)};
	$I->dbformat_out=$I->dbformat_in=
	  $dbformats{substr($dummy, FORMATVERSION, 4)};

	# read integer format
	$fmt=unpack 'x4a', $dummy;
	if( $fmt eq "\0" ) {
	  select undef, undef, undef, 0.1;
	  redo RETRY;
	}
	return unless $I->set_intfmt($fmt);

	# read the byte just after the format character
	$I->flags=unpack 'x5C', $dummy;

	$I->_data=\$dummy;		# now mapped

	# read main index position
	$I->mainidx=unpack('x'.(BASEOFFSET+MAINIDX*$I->_intsize).$I->intfmt,
			   ${$I->_data});
	$I->_ididx=unpack('x'.(BASEOFFSET+IDIDX*$I->_intsize).$I->intfmt,
			  ${$I->_data});
	$I->_stringtbl=unpack('x'.(BASEOFFSET+STRINGTBL*$I->_intsize).
			      $I->intfmt, ${$I->_data});

	# parameters: PARENT POS DATAMODE
	tie %{$I->main_index=+{}}, 'MMapDB::Index', $I, $I->mainidx, 0;
	tie %{$I->id_index=+{}}, 'MMapDB::IDIndex', $I, undef, 0;
      }
    }

  return $I;
}

sub stop {
  my ($I)=@_;

  $I->_e(E_TRANSACTION) if defined $I->_tmpfh;

  return $I unless defined $I->_data;

  for my $k (qw/_data _stringtbl mainidx _ididx/) {
    undef $I->$k;
  }

  untie %{$I->main_index}; undef $I->main_index;
  untie %{$I->id_index};   undef $I->id_index;

  return $I;
}

sub index_iterator {
  my ($I, $pos, $nth)=@_;

  my $data=$I->_data;
  return sub {} unless $data and defined $pos;
  my $fmt=$I->intfmt;
  my $isz=$I->_intsize;
  my ($nrecords, $recordlen)=unpack 'x'.$pos.$fmt.'2', $$data;
  die E_RANGE if $nth>$nrecords;
  $recordlen*=$isz;
  my ($cur, $end)=($pos+2*$isz+$nth*$recordlen,
		   $pos+2*$isz+$nrecords*$recordlen);
  my $stroff=$I->_stringtbl;
  my $sfmt=$I->_stringfmt;
  my $dbfmt=$I->dbformat_in;

  my $it=MMapDB::Iterator->new
    ( sub {
	if( @_ ) {
	  for( my $i=0; $i<@_; $i++ ) {
	    if( $_[$i]==IT_NTH ) {
	      my $nth=$_[++$i];
	      $nth=$pos+2*$isz+$nth*$recordlen;
	      die E_RANGE unless( $pos+2*$isz<=$nth and $nth<=$end );
	      $cur=$nth;
	      # return in VOID context
	      return unless defined wantarray;
	    } elsif( $_[$i]==IT_CUR ) {
	      return ($cur-2*$isz-$pos)/$recordlen;
	    } elsif( $_[$i]==IT_NELEM ) {
	      return ($end-2*$isz-$pos)/$recordlen;
	    }
	  }
	}
	return if $cur>=$end;
	my ($key, $npos)=unpack 'x'.$cur.$fmt.'2', $$data;
	my @list=unpack 'x'.($cur+2*$isz).$fmt.$npos, $$data;
	$cur+=$recordlen;
	if( $dbfmt>DBFMT0 ) {
	  my ($str, $utf8)=unpack('x'.($stroff+$key).$sfmt, $$data);
	  Encode::_utf8_on($str) if( $utf8 );
	  return ($str, @list);
	} else {
	  return (unpack('x'.($stroff+$key).$sfmt, $$data), @list);
	}
      }
    );

  return wantarray ? ($it, $nrecords) : $it;
}

sub id_index_iterator {
  my ($I)=@_;

  my $data=$I->_data;
  return sub {} unless $data;
  my $pos=$I->_ididx;
  my ($nrecords)=unpack 'x'.$pos.$I->intfmt, $$data;
  my $isz=$I->_intsize;
  my $recordlen=2*$isz;
  my ($cur, $end)=($pos+$isz,
		   $pos+$isz+$nrecords*$recordlen);
  my $fmt=$I->intfmt.'2';

  my $it=MMapDB::Iterator->new
    ( sub {
	if( @_ ) {
	  for( my $i=0; $i<@_; $i++ ) {
	    if( $_[$i]==IT_NTH ) {
	      my $nth=$_[++$i];
	      $nth=$pos+$isz+$nth*$recordlen;
	      die E_RANGE unless( $pos+$isz<=$nth and $nth<=$end );
	      $cur=$nth;
	      # return in VOID context
	      return unless defined wantarray;
	    } elsif( $_[$i]==IT_CUR ) {
	      return ($cur-$isz-$pos)/$recordlen;
	    } elsif( $_[$i]==IT_NELEM ) {
	      return ($end-$isz-$pos)/$recordlen;
	    }
	  }
	}
	return if $cur>=$end;
	my @l=unpack 'x'.$cur.$fmt, $$data;
	$cur+=$recordlen;
	return @l;
      }
    );

  return wantarray ? ($it, $nrecords) : $it;
}

sub is_datapos {
  my ($I, $pos)=@_;
  return $pos<$I->mainidx;
}

sub datamode : lvalue {
  tied(%{$_[0]->main_index})->datamode;
}

sub id_datamode : lvalue {
  tied(%{$_[0]->id_index})->datamode;
}

sub _e {$_[0]->_rollback; die $_[1]}
sub _ct {$_[0]->_tmpfh or die E_TRANSACTION}

sub begin {
  my ($I, $dbfmt)=@_;

  $I->_e(E_TRANSACTION) if defined $I->_tmpfh;

  die E_READONLY if $I->readonly;

  if( defined $I->lockfile ) {
    # open lockfile
    unless( ref $I->lockfile ) {
      open my $fh, '>', $I->lockfile or die E_OPEN;
      $I->lockfile=$fh;
    }
    flock $I->lockfile, LOCK_EX or die E_LOCK;
  }

  if (defined $dbfmt) {
    $I->dbformat_out=($dbfmt==-1 ? $#dbformats : $dbfmt);
  }
  $I->set_intfmt($I->intfmt);	# adjust string format

  $I->_tmpname=$I->filename.'.'.$$;

  {
    # open stringtbl tmpfile
    open my $fh, '+>', $I->_tmpname.'.strings' or die E_OPEN;
    $I->_stringfh=$fh;
    $I->_stringmap=\my $strings;
    eval {
      sysseek $fh, $I->stringmap_prealloc, SEEK_SET and
	truncate $fh, $I->stringmap_prealloc and
	  map_handle $strings, $fh, '+>', 0, $I->stringmap_prealloc;
    };
    die E_OPEN if $@;
    $I->_stringmap_end=0;
  }

  {
    # open tmpfile
    open my $fh, '+>', $I->_tmpname or die E_OPEN;
    $I->_tmpfh=$fh;		# this starts the transaction
    $I->_tmpmap=\do{my $map=''};

    $I->_putdata(0, 'a4aC', $dbformats[$I->dbformat_out], $I->intfmt,
		$I->flags & 0xff);
    $I->_index_end=BASEOFFSET+DATASTART*$I->_intsize;
  }

  # and copy every *valid* entry from the old file
  # create _idmap on the way
  $I->_idmap={};
  $I->_strpos=[];
  for( my $it=$I->iterator; my ($pos)=$it->(); ) {
    $I->insert($I->data_record($pos));
  }
  if( $I->_data ) {
    $I->_nextid=unpack('x'.(BASEOFFSET+NEXTID*$I->_intsize).$I->intfmt,
		       ${$I->_data});
  } else {
    $I->_nextid=1;
  }

  return $I;
}

# The interator() below hops over the mmapped area. This one works on the file.
# It can be used only within a begin/commit cycle.
sub _fiterator {
  my ($I, $end)=@_;

  my $map=$I->_tmpmap;
  my $pos=BASEOFFSET+$I->_intsize*DATASTART;

  return sub {
  LOOP: {
      return if $pos>=$end;
      my $elpos=$pos;

      # valid id nkeys key1...keyn sort data
      # read (valid, id, nkeys)
      my ($valid, $id, $nkeys)=unpack 'x'.$pos.$I->intfmt.'3', $$map;

      # move iterator position
      # 5: valid, id, nkeys ... sort, data
      $pos+=$I->_intsize*(5+$nkeys);
      redo LOOP unless ($valid);

      my @l=unpack 'x'.($elpos+3*$I->_intsize).$I->intfmt.($nkeys+2), $$map;
      my $data=pop @l;
      my $sort=pop @l;

      return ([\@l, $sort, $data, $id], $elpos);
    }
  };
}

sub _really_write_index {
  my ($I, $map, $level)=@_;

  my $recordlen=1;		# in ints: (1): for subindexes there is one
                                #          position to store

  # find the max. number of positions we have to store
  foreach my $v (values %$map) {
    if( ref($v) eq 'ARRAY' ) {
      # list of data records
      $recordlen=@$v if @$v>$recordlen;
    }
    # else: recordlen is initialized with 1. So for subindexes there is
    #       nothing to do
  }
  # each record comes with a header of 2 integers, the key position in the
  # string table and the actual position count of the record. So we have to
  # add 2 to $recordlen.
  $recordlen+=2;

  # the index itself has a 2 integer header, the recordlen and the number
  # of index records that belong to the index.
  my $indexsize=(2+$recordlen*keys(%$map))*$I->_intsize; # in bytes

  my $pos=$I->_index_end;

  # make room
  $I->_index_end=$pos+$indexsize;

  # and write subindices after this index
  my $strings=$I->_stringmap;
  my $sfmt=$I->_stringfmt_out;
  my $dbfmt=$I->dbformat_out;
  foreach my $v (values %$map) {
    if( ref($v) eq 'HASH' ) {
      # convert the subindex into a position list
      $v=[$I->_really_write_index($v, $level+1)];
    } else {
      # here we already have a position list but it still contains
      # sorting ids.
      @$v=map {
	$_->[1];
      } sort {
	$a->[0] cmp $b->[0];
      } map {
	# fetch sort string from string table
	if( $dbfmt>DBFMT0 ) {
	  my ($str, $utf8)=unpack('x'.$_->[0].$sfmt, $$strings);
	  Encode::_utf8_on($str) if $utf8;
	  [$str, $_->[1]];
	} else {
	  [unpack('x'.$_->[0].$sfmt, $$strings), $_->[1]];
	}
      } @$v;
    }
  }

  my $fmt=$I->intfmt;
  my $written=$pos;
  $written+=$I->_putdata($written, $fmt.'2', 0+keys(%$map), $recordlen);

  $fmt.=$recordlen;
  # write the records
  foreach my $key (map {
    $_->[0]
  } sort {
    $a->[1] cmp $b->[1];
  } map {
    if( $dbfmt>DBFMT0 ) {
      my ($str, $utf8)=unpack('x'.$_.$sfmt, $$strings);
      Encode::_utf8_on($str) if $utf8;
      [$_, $str];
    } else {
      [$_, unpack('x'.$_.$sfmt, $$strings)];
    }
  } keys %$map) {
    my $v=$map->{$key};

    #D($key, $v);
    #warn "$prefix> idx rec: ".unpack('H*', pack($fmt, $key, 0+@$v, @$v))."\n";

    $written+=$I->_putdata($written, $fmt, $key, 0+@$v, @$v);
  }

  return $pos;
}

sub _write_index {
  my ($I)=@_;

  my %map;
  for( my $it=$I->_fiterator($I->_index_end); my ($el, $pos)=$it->(); ) {
    my $m=\%map;
    my @k=@{$el->[0]};
    while(@k>1 and ref($m) eq 'HASH') {
      my $k=shift @k;
      $m->{$k}={} unless exists $m->{$k};
      $m=$m->{$k};
    }
    $I->_e(E_DUPLICATE) unless ref($m) eq 'HASH';
    $m->{$k[0]}=[] unless defined $m->{$k[0]};
    $I->_e(E_DUPLICATE) unless ref($m->{$k[0]}) eq 'ARRAY';
    # Actually we want to save only positions but they must be ordered.
    # So either keep the order field together with the position here to
    # sort it later or do sort of ordered insert here.
    # The former is simpler. So it's it.
    push @{$m->{$k[0]}}, [$el->[1], $pos];
  }

  return $I->_really_write_index(\%map, 0);
}

sub _write_id_index {
  my ($I)=@_;

  my $map=$I->_idmap;
  my $fmt=$I->intfmt;

  my $pos=$I->_index_end;
  my $written=$pos;
  $written+=$I->_putdata($written, $fmt, 0+keys(%$map));

  $fmt.='2';
  # write the records
  foreach my $key (sort {$a <=> $b} keys %$map) {
    my $v=$map->{$key};

    #warn "id> idx rec: ".unpack('H*', pack($fmt, $key, $v))."\n";

    $written+=$I->_putdata($written, $fmt, $key, $v);
  }
  $I->_index_end=$written;

  #warn sprintf "id> index written @ %#x\n", $pos;

  return $pos;
}

sub invalidate {
  my ($I)=@_;
  $I->_e(E_READONLY) if $I->readonly;
  return unless defined $I->_data;
  protect ${$I->_data}, '+<';
  substr( ${$I->_data}, INTFMT, 1, "\0" );
  protect ${$I->_data}, '<';
}

sub commit {
  my ($I, $dont_invalidate)=@_;

  $I->_ct;

  # write NEXTID
  $I->_putdata(BASEOFFSET+NEXTID*$I->_intsize, $I->intfmt, $I->_nextid);

  # write MAINIDX and IDIDX
  my $mainidx=$I->_write_index;
  my $ididx=$I->_write_id_index;

  $I->_putdata(BASEOFFSET+MAINIDX*$I->_intsize,   $I->intfmt, $mainidx);
  $I->_putdata(BASEOFFSET+IDIDX*$I->_intsize,     $I->intfmt, $ididx);
  $I->_putdata(BASEOFFSET+STRINGTBL*$I->_intsize, $I->intfmt, $I->_index_end);

  # now copy the string table
  my $fh=$I->_tmpfh;
  my $strings=$I->_stringmap;
  my $map=$I->_tmpmap;
  my $need=$I->_index_end+$I->_stringmap_end;
  if( $need>length $$map ) {
    eval {
      sysseek $fh, $need, SEEK_SET and
	truncate $fh, $need and
	  map_handle $$map, $fh, '+>', 0, $need;
    };
    $I->_e(E_OPEN) if $@;
  }
  substr($$map, $I->_index_end, $I->_stringmap_end,
	 substr($$strings, 0, $I->_stringmap_end));
  truncate $fh, $need;

  #warn "mainidx=$mainidx, ididx=$ididx, strtbl=$strtbl\n";

  undef $I->_idmap;
  undef $I->_strpos;
  undef $I->_stringmap;

  close $I->_stringfh or $I->_e(E_CLOSE);
  undef $I->_stringfh;
  unlink $I->_tmpname.'.strings';

  undef $I->_tmpmap;
  close $fh or $I->_e(E_CLOSE);
  undef $I->_tmpfh;

  # rename is (at least on Linux) an atomic operation
  rename $I->_tmpname, $I->filename  or $I->_e(E_RENAME);

  $I->invalidate unless $dont_invalidate;

  if( $I->lockfile ) {
    flock $I->lockfile, LOCK_UN or die E_LOCK;
  }

  $I->start;
}

sub _rollback {
  my ($I)=@_;

  close $I->_tmpfh;
  undef $I->_tmpfh;
  unlink $I->_tmpname;

  close $I->_stringfh;
  undef $I->_stringfh;
  unlink $I->_tmpname.'.strings';

  undef $I->_stringmap;
  undef $I->_strpos;
  undef $I->_idmap;

  $I->_stringfmt_out=$I->_stringfmt;
  $I->dbformat_out=$I->dbformat_in;

  if( $I->lockfile ) {
    flock $I->lockfile, LOCK_UN or die E_LOCK;
  }
}

sub rollback {
  my ($I)=@_;

  $I->_e(E_TRANSACTION) unless defined $I->_tmpfh;
  $I->_rollback;
}

sub DESTROY {
  my ($I)=@_;

  $I->_rollback if defined $I->_tmpfh;
  $I->stop;
}

sub backup {
  my ($I, $fn)=@_;

  $I->start;

  my $backup=$I->new(filename=>(defined $fn ? $fn : $I->filename.'.BACKUP'));

  $backup->begin;
  $backup->commit(1);
}

sub restore {
  my ($I, $fn)=@_;

  $I->start;
  $fn=$I->filename.'.BACKUP' unless defined $fn;

  # rename is (at least on Linux) an atomic operation
  rename $fn, $I->filename  or die E_RENAME;
  $I->invalidate;

  return $I->start;
}

# Returns the position of $key in the stringtable
# If $key is not found it is inserted. @{$I->_strpos} is kept ordered.
# So, we can do a binary search.
# This implements something very similar to a HASH. So, why not use a HASH?
# A HASH is kept completely in core and the memory is not returned to the
# operating system when finished. The number of strings in the database
# can become quite large. So if a long running process updates the database
# only once it will consume much memory for nothing. To avoid this we map
# the string table currently under construction in a separate file that
# is mmapped into the address space of this process and keep here only
# a list of pointer into this area. When the transaction is committed the
# memory is returned to the OS. But on the other hand we need fast access.
# This is achieved by the binary search.
sub _string2pos {
  my ($I, $key)=@_;

  my $fmt=$I->_stringfmt_out;
  my $dbfmt=$I->dbformat_out;

  Encode::_utf8_off($key) if $dbfmt==DBFMT0;

  my $strings=$I->_stringmap;
  my $poslist=$I->_strpos;

  my ($low, $high)=(0, 0+@$poslist);
  #warn "_string2pos($key): low=$low, high=$high\n";

  my ($cur, $rel, $curstr);
  while( $low<$high ) {
    $cur=($high+$low)/2;	# "use integer" is active, see above
    if( $dbfmt>DBFMT0 ) {
      ($curstr, my $utf8)=unpack 'x'.$poslist->[$cur].$fmt, $$strings;
      Encode::_utf8_on($curstr) if $utf8;
    } else {
      $curstr=unpack 'x'.$poslist->[$cur].$fmt, $$strings;
    }
    #warn "  --> looking at $curstr: low=$low, high=$high, cur=$cur\n";
    $rel=($curstr cmp $key);
    if( $rel<0 ) {
      #warn "  --> moving low: $low ==> ".($cur+1)."\n";
      $low=$cur+1;
    } elsif( $rel>0 ) {
      #warn "  --> moving high: $high ==> ".($cur)."\n";
      # don't try to optimize here: $high=$cur-1 will not work in border cases
      $high=$cur;
    } else {
      #warn "  --> BINGO\n";
      return $poslist->[$cur];
    }
  }
  #warn "  --> NOT FOUND\n";
  my $pos=$I->_stringmap_end;
  splice @$poslist, $low, 0, $pos;
  #warn "  --> inserting $pos into poslist at $low ==> @$poslist\n";

  my $newstr;
  if( $dbfmt>DBFMT0 ) {
    if( Encode::is_utf8($key) ) {
      $newstr=pack($fmt, Encode::encode_utf8($key), 1);
    } else {
      $newstr=pack($fmt, $key, 0);
    }
  } else {
    $newstr=pack($fmt, $key);
  }

  if( $pos+length($newstr)>length $$strings ) {
    # remap
    my $prea=$I->stringmap_prealloc;
    my $need=$prea*(($pos+length($newstr)+$prea-1)/$prea);
    eval {
      my $fh=$I->_stringfh;
      sysseek $fh, $need, SEEK_SET and
	truncate $fh, $need and
	  map_handle $$strings, $fh, '+>', 0, $need;
    };
    $I->_e(E_OPEN) if $@;
  }

  substr $$strings, $pos, length($newstr), $newstr;
  $I->_stringmap_end=$pos+length($newstr);

  return $pos;
}

sub insert {
  my ($I, $rec)=@_;
  #my ($I, $key, $sort, $data, $id)=@_;

  #use Data::Dumper; $Data::Dumper::Useqq=1; warn Dumper $rec;

  $I->_ct;

  $rec->[0]=[$rec->[0]] unless ref $rec->[0];
  for my $v (@{$rec}[1,2]) {
    $v='' unless defined $v;
  }

  # create new ID if necessary
  my $id=$rec->[3];
  my $idmap=$I->_idmap;
  if( defined $id ) {
    $I->_e(E_TWICE) if exists $idmap->{$id};
  } else {
    $id=$I->_nextid;
    undef $idmap->{$id};	# allocate it

    my $mask=do{no integer; unpack( $I->intfmt, pack $I->intfmt, -1 )>>1};
    my $nid=($id+1)&$mask;
    $nid=1 if $nid==0;
    while(exists $idmap->{$nid}) {
      $nid=($nid+1)&$mask; $nid=1 if $nid==0;
      $I->_e(E_FULL) if $nid==$id;
    }
    $I->_nextid=$nid;
  }

  my $pos=$I->_index_end;
  $I->_index_end+=$I->_putdata($pos, $I->intfmt.'*', 1, $id, 0+@{$rec->[0]},
			      map {$I->_string2pos($_)}
			      @{$rec->[0]}, @{$rec}[1,2]);

  $idmap->{$id}=$pos;

  return ($id, $pos);
}

sub delete_by_id {
  my ($I, $id, $return_element)=@_;

#   warn "delete_by_id($id)\n";

  # no such id
  return unless exists $I->_idmap->{$id};

  my $map=$I->_tmpmap;
  my $idmap=$I->_idmap;
  my $pos;

  return unless defined($pos=delete $idmap->{$id});

  # read VALID, ID, NKEYS
  my ($valid, $elid, $nkeys)=unpack 'x'.$pos.$I->intfmt.'3', $$map;

  return unless $valid;
  return unless $id==$elid; # XXX: should'nt that be an E_CORRUPT

  my $rc=1;
  if( $return_element ) {
    my $strings=$I->_stringmap;
    my $sfmt=$I->_stringfmt_out;
    my $dbfmt=$I->dbformat_out;
    my @l=map {
      if( $dbfmt>DBFMT0 ) {
	my ($str, $utf8)=unpack('x'.$_.$sfmt, $$strings);
	Encode::_utf8_on($str) if $utf8;
	$str;
      } else {
	unpack('x'.$_.$sfmt, $$strings);
      }
    } unpack('x'.($pos+3*$I->_intsize).$I->intfmt.($nkeys+2), $$map);

    my $rdata=pop @l;
    my $rsort=pop @l;

    $rc=[\@l, $rsort, $rdata, $id];
  }

  $I->_putdata($pos, $I->intfmt, 0); # invalidate the record

  return $rc;
}

sub clear {
  my ($I)=@_;

  $I->_ct;

  $I->_index_end=BASEOFFSET+DATASTART*$I->_intsize;
  $I->_stringmap_end=0;

  $I->_idmap={};
  $I->_strpos=[];

  return;
}

# sub xdata_record {
#   my ($I, $pos)=@_;

#   return unless $pos>0 and $pos<$I->mainidx;

#   # valid id nkeys key1...keyn sort data
#   my ($id, $nkeys)=unpack('x'.($pos+$I->_intsize).' '.$I->intfmt.'3',
# 			  ${$I->_data});

#   my $off=$I->_stringtbl;
#   my $data=$I->_data;
#   my $sfmt=$I->_stringfmt;
#   my @l=map {
#     unpack('x'.($off+$_).$sfmt, $$data);
#   } unpack('x'.($pos+3*$I->_intsize).' '.$I->intfmt.($nkeys+2), $$data);

#   my $rdata=pop @l;
#   my $rsort=pop @l;

#   #warn "data_record: keys=[@l], sort=$rsort, data=$rdata, id=$id\n";
#   return [\@l, $rsort, $rdata, $id];
# }

sub iterator {
  my ($I, $show_invalid)=@_;

  return sub {} unless $I->_data;

  my $pos=BASEOFFSET+DATASTART*$I->_intsize;
  my $end=$I->mainidx;

  return MMapDB::Iterator->new
    (sub {
       die E_NOT_IMPLEMENTED if @_;
     LOOP: {
	 return if $pos>=$end;

	 # valid id nkeys key1...keyn sort data
	 my ($valid, undef, $nkeys)=
	   unpack 'x'.$pos.' '.$I->intfmt.'3', ${$I->_data};

	 if( $valid xor $show_invalid ) {
	   my $rc=$pos;
	   $pos+=$I->_intsize*($nkeys+5); # 5=(valid id nkeys sort data)
	   return $rc;
	 }
	 $pos+=$I->_intsize*($nkeys+5); # 5=(valid id nkeys sort data)
	 redo LOOP;
       }
     });
}

package MMapDB::Iterator;

use strict;

sub new {
  my ($class, $func)=@_;
  $class=ref($class) || $class;
  return bless $func=>$class;
}

sub nth {
  return $_[0]->(MMapDB::IT_NTH, $_[1]);
}

sub cur {
  return $_[0]->(MMapDB::IT_CUR);
}

sub nelem {
  return $_[0]->(MMapDB::IT_NELEM);
}

#######################################################################
# High Level Accessor Classes
#######################################################################

{
  package
    MMapDB::_base;

  use strict;
  use Carp qw/croak/;
  use Scalar::Util ();
  use Exporter qw/import/;

  use constant ({
                 PARENT=>0,
                 POS=>1,
                 DATAMODE=>2,
                 ITERATOR=>3,
                 SHADOW=>4,
                });
  BEGIN {our @EXPORT=(qw!PARENT POS DATAMODE ITERATOR SHADOW!)};

  sub new {
    my ($class, @param)=@_;
    $class=ref($class) || $class;
    $param[DATAMODE]=0 unless defined $param[DATAMODE];
    Scalar::Util::weaken $param[0];
    return bless \@param=>$class;
  }

  sub readonly {croak "Modification of a read-only value attempted";}

  sub datamode : lvalue {$_[0]->[DATAMODE]}

  BEGIN {
    *TIEHASH=\&new;
    # STORE must be allowed to support constructs like this (with aliases):
    #   map {
    #     local $_;
    #   } values %{$db->main_index};
    # or
    #   for (values %{$db->main_index}) {
    #     local $_;
    #   }
    *STORE=sub {
      my ($I, $key, $value)=@_;
      my $el;
      my $ll=MMapDB::_localizing();
      # Carp::cluck "PL_localizing=$ll";

      $el=($I->[SHADOW]||={});
      my $sh;
      if( $ll==0 and $sh=$el->{$key} ) { # is already localized
        # warn "  ==> already shadowed";
        $sh->[1]=$value;
      } elsif( $ll==1 ) {
        # warn "  ==> shadowing";
        $sh=($el->{$key}||=[]);
        $sh->[0]++;
        $sh->[1]=$value;
      } elsif( $ll==2 ) {
        if( --$sh->[0] ) {
          # warn "  ==> decremented shadow counter";
          $sh->[1]=$value;
        } else {
          # warn "  ==> deleting shadow";
          delete $el->{$key};
        }
      } else {
        # warn "  ==> ro";
        goto &readonly;
      }
    };
    *DELETE=\&readonly;
    *CLEAR=\&readonly;

    *TIEARRAY=\&new;
    #*STORE=sub {};
    *STORESIZE=\&readonly;
    *EXTEND=\&readonly;
    #*DELETE=\&readonly;
    #*CLEAR=\&readonly;
    *PUSH=\&readonly;
    *UNSHIFT=\&readonly;
    *POP=\&readonly;
    *SHIFT=\&readonly;
    *SPLICE=\&readonly;
  }
}

#######################################################################
# Normal Index Accessor
#######################################################################

{
  package MMapDB::Index;

  use strict;
  BEGIN {MMapDB::_base->import}
  {our @ISA=qw/MMapDB::_base/}

  sub FETCH {
    my ($I, $key)=@_;

    {
      my $shel;
      $shel=$I->[SHADOW] and
        keys %$shel and
          $shel=$shel->{$key} and
            return $shel->[1];
    }

    my @el=$I->[PARENT]->index_lookup($I->[POS], $key);

    return unless @el;

    my $rc;

    if( @el==1 and $el[0]>=$I->[PARENT]->mainidx ) {
      # another index
      tie %{$rc={}}, ref($I), $I->[PARENT], $el[0], $I->[DATAMODE];
    } else {
      tie @{$rc=[]}, 'MMapDB::Data', $I->[PARENT], \@el, $I->[DATAMODE];
    }

    return $rc;
  }

  sub EXISTS {
    my ($I, $key)=@_;
    return $I->[PARENT]->index_lookup($I->[POS], $key) ? 1 : undef;
  }

  sub FIRSTKEY {
    my ($I)=@_;
    my @el=($I->[ITERATOR]=$I->[PARENT]->index_iterator($I->[POS]))->();
    return @el ? $el[0] : ();
  }

  sub NEXTKEY {
    my ($I)=@_;
    my @el=$I->[ITERATOR]->();
    return @el ? $el[0] : ();
  }

  sub SCALAR {
    my ($I)=@_;
    my $pos=defined $I->[POS] ? $I->[POS] : $I->[PARENT]->_ididx;
    my $n=unpack 'x'.$pos.$I->[PARENT]->intfmt,${$I->[PARENT]->_data};
    return $n==0 ? $n : "$n/$n";
  }
}

#######################################################################
# ID Index Accessor
#######################################################################

{
  package MMapDB::IDIndex;

  use strict;
  BEGIN {MMapDB::_base->import}
  {our @ISA=qw/MMapDB::Index/}

  sub FETCH {
    {
      my $shel;
      $shel=$_[0]->[SHADOW] and
        keys %$shel and
          $shel=$shel->{$_[1]} and
            return $shel->[1];
    }

    if( $_[0]->[DATAMODE]==MMapDB::DATAMODE_SIMPLE ) {
      $_[0]->[PARENT]->data_value($_[0]->[PARENT]->id_index_lookup($_[1]));
    } else {
      $_[0]->[PARENT]->data_record($_[0]->[PARENT]->id_index_lookup($_[1]));
    }
  }

  sub EXISTS {
    my ($I, $key)=@_;
    return $I->[PARENT]->id_index_lookup($key) ? 1 : undef;
  }

  sub FIRSTKEY {
    my ($I)=@_;
    my @el=($I->[ITERATOR]=$I->[PARENT]->id_index_iterator)->();
    return @el ? $el[0] : ();
  }
}

#######################################################################
# Data Accessor
#######################################################################

{
  package MMapDB::Data;

  use strict;
  BEGIN {MMapDB::_base->import}
  {our @ISA=qw/MMapDB::_base/}

  sub FETCH {
    my ($I, $idx)=@_;

    {
      my $shel;
      $shel=$I->[SHADOW] and
        keys %$shel and
          $shel=$shel->{$idx} and
            return $shel->[1];
    }

    return unless @{$I->[POS]}>$idx;
    if( $I->[DATAMODE]==MMapDB::DATAMODE_SIMPLE ) {
      return $I->[PARENT]->data_value($I->[POS]->[$idx]);
    } else {
      return $I->[PARENT]->data_record($I->[POS]->[$idx]);
    }
  }

  sub FETCHSIZE {scalar @{$_[0]->[POS]}}

  sub EXISTS {@{$_[0]->[POS]}>$_[1]}
}

1;
__END__