[dbsrgits/DBM-Deep.git] / lib / DBM / Deep / Engine3.pm

package DBM::Deep::Engine3;

use 5.6.0;

use strict;

our $VERSION = q(0.99_03);

use Scalar::Util ();

# File-wide notes:
# * Every method in here assumes that the storage has been appropriately
#   safeguarded. This can be anything from flock() to some sort of manual
#   mutex. But, it's the caller's responsability to make sure that this has
#   been done.

# Setup file and tag signatures.  These should never change.
sub SIG_FILE     () { 'DPDB' }
sub SIG_HEADER   () { 'h'    }
sub SIG_INTERNAL () { 'i'    }
sub SIG_HASH     () { 'H'    }
sub SIG_ARRAY    () { 'A'    }
sub SIG_NULL     () { 'N'    }
sub SIG_DATA     () { 'D'    }
sub SIG_INDEX    () { 'I'    }
sub SIG_BLIST    () { 'B'    }
sub SIG_FREE     () { 'F'    }
sub SIG_KEYS     () { 'K'    }
sub SIG_SIZE     () {  1     }

################################################################################

# Please refer to the pack() documentation for further information
my %StP = (
    1 => 'C', # Unsigned char value
    2 => 'n', # Unsigned short in "network" (big-endian) order
    4 => 'N', # Unsigned long in "network" (big-endian) order
    8 => 'Q', # Usigned quad (no order specified, presumably machine-dependent)
);

sub new {
    my $class = shift;
    my ($args) = @_;

    my $self = bless {
        byte_size   => 4,

        digest      => undef,
        hash_size   => 16, # In bytes
        max_buckets => 16,
        num_txns    => 16, # HEAD plus 15 running txns
        trans_id    => 0,  # Default to the HEAD

        entries => {}, # This is the list of entries for transactions
        storage => undef,
    }, $class;

    if ( defined $args->{pack_size} ) {
        if ( lc $args->{pack_size} eq 'small' ) {
            $args->{byte_size} = 2;
        }
        elsif ( lc $args->{pack_size} eq 'medium' ) {
            $args->{byte_size} = 4;
        }
        elsif ( lc $args->{pack_size} eq 'large' ) {
            $args->{byte_size} = 8;
        }
        else {
            die "Unknown pack_size value: '$args->{pack_size}'\n";
        }
    }

    # Grab the parameters we want to use
    foreach my $param ( keys %$self ) {
        next unless exists $args->{$param};
        $self->{$param} = $args->{$param};
    }

    $self->{byte_pack} = $StP{ $self->byte_size };

    ##
    # Number of buckets per blist before another level of indexing is
    # done. Increase this value for slightly greater speed, but larger database
    # files. DO NOT decrease this value below 16, due to risk of recursive
    # reindex overrun.
    ##
    if ( $self->{max_buckets} < 16 ) {
        warn "Floor of max_buckets is 16. Setting it to 16 from '$self->{max_buckets}'\n";
        $self->{max_buckets} = 16;
    }

    if ( !$self->{digest} ) {
        require Digest::MD5;
        $self->{digest} = \&Digest::MD5::md5;
    }

    return $self;
}

################################################################################

sub read_value {
    my $self = shift;
    my ($obj, $key) = @_;

    # This will be a Reference sector
    my $sector = $self->_load_sector( $obj->_base_offset )
        or die "How did read_value fail (no sector for '$obj')?!\n";

    my $key_md5 = $self->_apply_digest( $key );

    my $value_sector = $sector->get_data_for({
        key_md5    => $key_md5,
        allow_head => 1,
    });

    unless ( $value_sector ) {
        $value_sector = DBM::Deep::Engine::Sector::Null->new({
            engine => $self,
            data   => undef,
        });

        $sector->write_data({
            key_md5 => $key_md5,
            key     => $key,
            value   => $value_sector,
        });
    }

    return $value_sector->data;
}

sub get_classname {
    my $self = shift;
    my ($obj) = @_;

    # This will be a Reference sector
    my $sector = $self->_load_sector( $obj->_base_offset )
        or die "How did read_value fail (no sector for '$obj')?!\n";

    return $sector->get_classname;
}

sub key_exists {
    my $self = shift;
    my ($obj, $key) = @_;

    # This will be a Reference sector
    my $sector = $self->_load_sector( $obj->_base_offset )
        or die "How did key_exists fail (no sector for '$obj')?!\n";

    my $data = $sector->get_data_for({
        key_md5    => $self->_apply_digest( $key ),
        allow_head => 1,
    });

    # exists() returns 1 or '' for true/false.
    return $data ? 1 : '';
}

sub delete_key {
    my $self = shift;
    my ($obj, $key) = @_;

    my $sector = $self->_load_sector( $obj->_base_offset )
        or die "How did delete_key fail (no sector for '$obj')?!\n";

    return $sector->delete_key({
        key_md5    => $self->_apply_digest( $key ),
        allow_head => 0,
    });
}

sub write_value {
    my $self = shift;
    my ($obj, $key, $value) = @_;

    my $r = Scalar::Util::reftype( $value ) || '';
    {
        last if $r eq '';
        last if $r eq 'HASH';
        last if $r eq 'ARRAY';

        DBM::Deep->_throw_error(
            "Storage of references of type '$r' is not supported."
        );
    }

    my ($class, $type);
    if ( !defined $value ) {
        $class = 'DBM::Deep::Engine::Sector::Null';
    }
    elsif ( $r eq 'ARRAY' || $r eq 'HASH' ) {
        if ( $r eq 'ARRAY' && tied(@$value) ) {
            DBM::Deep->_throw_error( "Cannot store something that is tied." );
        }
        if ( $r eq 'HASH' && tied(%$value) ) {
            DBM::Deep->_throw_error( "Cannot store something that is tied." );
        }
        $class = 'DBM::Deep::Engine::Sector::Reference';
        $type = substr( $r, 0, 1 );
    }
    else {
        $class = 'DBM::Deep::Engine::Sector::Scalar';
    }

    # This will be a Reference sector
    my $sector = $self->_load_sector( $obj->_base_offset )
        or die "How did write_value fail (no sector for '$obj')?!\n";

    # Create this after loading the reference sector in case something bad happens.
    # This way, we won't allocate value sector(s) needlessly.
    my $value_sector = $class->new({
        engine => $self,
        data   => $value,
        type   => $type,
    });

    $sector->write_data({
        key     => $key,
        key_md5 => $self->_apply_digest( $key ),
        value   => $value_sector,
    });

    # This code is to make sure we write all the values in the $value to the disk
    # and to make sure all changes to $value after the assignment are reflected
    # on disk. This may be counter-intuitive at first, but it is correct dwimmery.
    #   NOTE - simply tying $value won't perform a STORE on each value. Hence, the
    # copy to a temp value.
    if ( $r eq 'ARRAY' ) {
        my @temp = @$value;
        tie @$value, 'DBM::Deep', {
            base_offset => $value_sector->offset,
            storage     => $self->storage,
            engine      => $self,
        };
        @$value = @temp;
        bless $value, 'DBM::Deep::Array' unless Scalar::Util::blessed( $value );
    }
    elsif ( $r eq 'HASH' ) {
        my %temp = %$value;
        tie %$value, 'DBM::Deep', {
            base_offset => $value_sector->offset,
            storage     => $self->storage,
            engine      => $self,
        };

        %$value = %temp;
        bless $value, 'DBM::Deep::Hash' unless Scalar::Util::blessed( $value );
    }

    return 1;
}

sub get_next_key {
    my $self = shift;
    my ($obj, $prev_key) = @_;

    # XXX Need to add logic about resetting the iterator if any key in the reference has changed
    unless ( $prev_key ) {
        $obj->{iterator} = DBM::Deep::Engine::Iterator->new({
            base_offset => $obj->_base_offset,
            engine      => $self,
        });
    }

    return $obj->{iterator}->get_next_key;
}

################################################################################

sub setup_fh {
    my $self = shift;
    my ($obj) = @_;

    # We're opening the file.
    unless ( $obj->_base_offset ) {
        my $bytes_read = $self->_read_file_header;

        # Creating a new file
        unless ( $bytes_read ) {
            $self->_write_file_header;

            # 1) Create Array/Hash entry
            my $initial_reference = DBM::Deep::Engine::Sector::Reference->new({
                engine => $self,
                type   => $obj->_type,
            });
            $obj->{base_offset} = $initial_reference->offset;

            $self->storage->flush;
        }
        # Reading from an existing file
        else {
            $obj->{base_offset} = $bytes_read;
            my $initial_reference = DBM::Deep::Engine::Sector::Reference->new({
                engine => $self,
                offset => $obj->_base_offset,
            });
            unless ( $initial_reference ) {
                DBM::Deep->_throw_error("Corrupted file, no master index record");
            }

            unless ($obj->_type eq $initial_reference->type) {
                DBM::Deep->_throw_error("File type mismatch");
            }
        }
    }

    return 1;
}

sub begin_work {
    my $self = shift;
    my ($obj) = @_;

    if ( $self->trans_id ) {
        DBM::Deep->_throw_error( "Cannot begin_work within a transaction" );
    }

    my @slots = $self->read_txn_slots;
    for my $i ( 1 .. @slots ) {
        next if $slots[$i];
        $slots[$i] = 1;
        $self->set_trans_id( $i );
        last;
    }
    $self->write_txn_slots( @slots );

    if ( !$self->trans_id ) {
        DBM::Deep->_throw_error( "Cannot begin_work - no available transactions" );
    }

    $self->set_trans_ctr(
        $self->get_txn_staleness_counter( $self->trans_id ),
    );

    return;
}

sub rollback {
    my $self = shift;
    my ($obj) = @_;

    if ( !$self->trans_id ) {
        DBM::Deep->_throw_error( "Cannot rollback without a transaction" );
    }

    # Each entry is the file location for a bucket that has a modification for
    # this transaction. The entries need to be expunged.
    foreach my $entry (@{ $self->get_entries } ) {
        # Remove the entry here
        my $read_loc = $entry
          + $self->hash_size
          + $self->byte_size
          + $self->trans_id * ( $self->byte_size + 4 );

        my $data_loc = $self->storage->read_at( $read_loc, $self->byte_size );
        $data_loc = unpack( $StP{$self->byte_size}, $data_loc );
        $self->storage->print_at( $read_loc, pack( $StP{$self->byte_size}, 0 ) );

        if ( $data_loc > 1 ) {
            $self->_load_sector( $data_loc )->free;
        }
    }

    $self->clear_entries;

    my @slots = $self->read_txn_slots;
    $slots[$self->trans_id] = 0;
    $self->write_txn_slots( @slots );
    $self->inc_txn_staleness_counter( $self->trans_id );
    $self->set_trans_ctr( 0 );
    $self->set_trans_id( 0 );

    return 1;
}

sub commit {
    my $self = shift;
    my ($obj) = @_;

    if ( !$self->trans_id ) {
        DBM::Deep->_throw_error( "Cannot commit without a transaction" );
    }

    #print "TID: " . $self->trans_id, $/;
    foreach my $entry (@{ $self->get_entries } ) {
        #print "$entry\n";
        # Overwrite the entry in head with the entry in trans_id
        my $base = $entry
          + $self->hash_size
          + $self->byte_size;

        my $head_loc = $self->storage->read_at( $base, $self->byte_size );
        $head_loc = unpack( $StP{$self->byte_size}, $head_loc );
        my $trans_loc = $self->storage->read_at(
            $base + $self->trans_id * ( $self->byte_size + 4 ), $self->byte_size,
        );

        $self->storage->print_at( $base, $trans_loc );
        $self->storage->print_at(
            $base + $self->trans_id * ( $self->byte_size + 4 ),
            pack( $StP{$self->byte_size} . ' N', (0) x 2 ),
        );

        if ( $head_loc > 1 ) {
            $self->_load_sector( $head_loc )->free;
        }
    }

    $self->clear_entries;

    my @slots = $self->read_txn_slots;
    $slots[$self->trans_id] = 0;
    $self->write_txn_slots( @slots );
    $self->inc_txn_staleness_counter( $self->trans_id );
    $self->set_trans_ctr( 0 );
    $self->set_trans_id( 0 );

    return 1;
}

sub read_txn_slots {
    my $self = shift;
    return split '', unpack( 'b32',
        $self->storage->read_at(
            $self->trans_loc, 4,
        )
    );
}

sub write_txn_slots {
    my $self = shift;
    $self->storage->print_at( $self->trans_loc,
        pack( 'b32', join('', @_) ),
    );
}

sub get_running_txn_ids {
    my $self = shift;
    my @transactions = $self->read_txn_slots;
    my @trans_ids = grep { $transactions[$_] } 0 .. $#transactions;
}

sub get_txn_staleness_counter {
    my $self = shift;
    my ($trans_id) = @_;
    return unpack( 'N',
        $self->storage->read_at(
            $self->trans_loc + 4 * ($trans_id - 1), 4,
        )
    );
}

sub inc_txn_staleness_counter {
    my $self = shift;
    my ($trans_id) = @_;
    $self->storage->print_at(
        $self->trans_loc,
        pack( 'N', $self->get_txn_staleness_counter( $trans_id ) + 1 ),
    );
}

sub get_entries {
    my $self = shift;
    return [ keys %{ $self->{entries}{$self->trans_id} ||= {} } ];
}

sub add_entry {
    my $self = shift;
    my ($trans_id, $loc) = @_;

    #print "$trans_id => $loc\n";
    $self->{entries}{$trans_id} ||= {};
    $self->{entries}{$trans_id}{$loc} = undef;
    #use Data::Dumper;print "$self: " . Dumper $self->{entries};
}

sub clear_entries {
    my $self = shift;
    #print "Clearing\n";
    delete $self->{entries}{$self->trans_id};
}

################################################################################

{
    my $header_fixed = length( SIG_FILE ) + 1 + 4 + 4;

    sub _write_file_header {
        my $self = shift;

        my $header_var = 1 + 1 + 4 + 4 * $self->num_txns + 2 * $self->byte_size;

        my $loc = $self->storage->request_space( $header_fixed + $header_var );

        $self->storage->print_at( $loc,
            SIG_FILE,
            SIG_HEADER,
            pack('N', 1),           # header version - at this point, we're at 9 bytes
            pack('N', $header_var), # header size
            # --- Above is $header_fixed. Below is $header_var
            pack('C', $self->byte_size),
            pack('C', $self->max_buckets),
            pack('N', 0 ),                   # Transaction cctiveness bitfield
            pack('N' . $self->num_txns, 0 ), # Transaction staleness counters
            pack($StP{$self->byte_size}, 0), # Start of free chain (blist size)
            pack($StP{$self->byte_size}, 0), # Start of free chain (data size)
        );

        $self->set_trans_loc( $header_fixed + 2 );
        $self->set_chains_loc( $header_fixed + 2 + 4 + 4 * $self->num_txns );
        $self->set_trans_ctr( 0 );

        return;
    }

    sub _read_file_header {
        my $self = shift;

        my $buffer = $self->storage->read_at( 0, $header_fixed );
        return unless length($buffer);

        my ($file_signature, $sig_header, $header_version, $size) = unpack(
            'A4 A N N', $buffer
        );

        unless ( $file_signature eq SIG_FILE ) {
            $self->storage->close;
            DBM::Deep->_throw_error( "Signature not found -- file is not a Deep DB" );
        }

        unless ( $sig_header eq SIG_HEADER ) {
            $self->storage->close;
            DBM::Deep->_throw_error( "Old file version found." );
        }

        my $buffer2 = $self->storage->read_at( undef, $size );
        my @values = unpack( 'C C', $buffer2 );

        $self->set_trans_loc( $header_fixed + 2 );
        $self->set_chains_loc( $header_fixed + 2 + 4 + 4 * $self->num_txns );
        $self->set_trans_ctr( 0 );

        if ( @values < 2 || grep { !defined } @values ) {
            $self->storage->close;
            DBM::Deep->_throw_error("Corrupted file - bad header");
        }

        #XXX Add warnings if values weren't set right
        @{$self}{qw(byte_size max_buckets)} = @values;

        my $header_var = 1 + 1 + 4 + 4 * $self->num_txns + 2 * $self->byte_size;
        unless ( $size eq $header_var ) {
            $self->storage->close;
            DBM::Deep->_throw_error( "Unexpected size found ($size <-> $header_var)." );
        }

        return length($buffer) + length($buffer2);
    }
}

sub _load_sector {
    my $self = shift;
    my ($offset) = @_;

    my $type = $self->storage->read_at( $offset, 1 );
    return if $type eq chr(0);

    if ( $type eq $self->SIG_ARRAY || $type eq $self->SIG_HASH ) {
        return DBM::Deep::Engine::Sector::Reference->new({
            engine => $self,
            type   => $type,
            offset => $offset,
        });
    }
    # XXX Don't we need key_md5 here?
    elsif ( $type eq $self->SIG_BLIST ) {
        return DBM::Deep::Engine::Sector::BucketList->new({
            engine => $self,
            type   => $type,
            offset => $offset,
        });
    }
    elsif ( $type eq $self->SIG_NULL ) {
        return DBM::Deep::Engine::Sector::Null->new({
            engine => $self,
            type   => $type,
            offset => $offset,
        });
    }
    elsif ( $type eq $self->SIG_DATA ) {
        return DBM::Deep::Engine::Sector::Scalar->new({
            engine => $self,
            type   => $type,
            offset => $offset,
        });
    }
    # This was deleted from under us, so just return and let the caller figure it out.
    elsif ( $type eq $self->SIG_FREE ) {
        return;
    }

    die "'$offset': Don't know what to do with type '$type'\n";
}

sub _apply_digest {
    my $self = shift;
    return $self->{digest}->(@_);
}

sub _add_free_sector {
    my $self = shift;
    my ($offset, $size) = @_;

    my $chains_offset;
    # Data sector
    if ( $size == 256 ) {
        $chains_offset = $self->byte_size;
    }
    # Blist sector
    else {
        $chains_offset = 0;
    }

    my $old_head = $self->storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );

    $self->storage->print_at( $self->chains_loc + $chains_offset,
        pack( $StP{$self->byte_size}, $offset ),
    );

    # Record the old head in the new sector after the signature
    $self->storage->print_at( $offset + 1, $old_head );
}

sub _request_sector {
    my $self = shift;
    my ($size) = @_;

    my $chains_offset;
    # Data sector
    if ( $size == 256 ) {
        $chains_offset = $self->byte_size;
    }
    # Blist sector
    else {
        $chains_offset = 0;
    }

    my $old_head = $self->storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
    my $loc = unpack( $StP{$self->byte_size}, $old_head );

    # We don't have any free sectors of the right size, so allocate a new one.
    unless ( $loc ) {
        return $self->storage->request_space( $size );
    }

    my $new_head = $self->storage->read_at( $loc + 1, $self->byte_size );
    $self->storage->print_at( $self->chains_loc + $chains_offset, $new_head );

    return $loc;
}

################################################################################

sub storage     { $_[0]{storage} }
sub byte_size   { $_[0]{byte_size} }
sub hash_size   { $_[0]{hash_size} }
sub num_txns    { $_[0]{num_txns} }
sub max_buckets { $_[0]{max_buckets} }
sub blank_md5   { chr(0) x $_[0]->hash_size }

sub trans_id     { $_[0]{trans_id} }
sub set_trans_id { $_[0]{trans_id} = $_[1] }

sub trans_ctr     { $_[0]{trans_ctr} }
sub set_trans_ctr { $_[0]{trans_ctr} = $_[1] }

sub trans_loc     { $_[0]{trans_loc} }
sub set_trans_loc { $_[0]{trans_loc} = $_[1] }

sub chains_loc     { $_[0]{chains_loc} }
sub set_chains_loc { $_[0]{chains_loc} = $_[1] }

################################################################################

package DBM::Deep::Engine::Iterator;

sub new {
    my $class = shift;
    my ($args) = @_;

    my $self = bless {
        breadcrumbs => [],
        engine      => $args->{engine},
        base_offset => $args->{base_offset},
    }, $class;

    Scalar::Util::weaken( $self->{engine} );

    return $self;
}

sub reset {
    my $self = shift;
    $self->{breadcrumbs} = [];
}

sub get_next_key {
    my $self = shift;

    my $crumbs = $self->{breadcrumbs};

    unless ( @$crumbs ) {
        # This will be a Reference sector
        my $sector = $self->{engine}->_load_sector( $self->{base_offset} )
            # or die "Iterator: How did this fail (no ref sector for '$self->{base_offset}')?!\n";
            # If no sector is found, thist must have been deleted from under us.
            or return;
        push @$crumbs, [ $sector->get_blist_loc, 0 ];
    }

    my $key;
    while ( 1 ) {
        my ($offset, $idx) = @{ $crumbs->[-1] };
        unless ( $offset ) {
            $self->reset;
            last;
        }

        if ( $idx >= $self->{engine}->max_buckets ) {
            $self->reset;
            last;
        }

        my $sector = $self->{engine}->_load_sector( $offset )
            or die "Iterator: How did this fail (no blist sector for '$offset')?!\n";

        #XXX Think this through!
        my $loc =  $sector->get_data_location_for({
            idx => $idx,
            allow_head => 1,
        });
        unless ( $loc ) {
            $crumbs->[-1][1]++;
            next;
        }

        my $key_sector = $sector->get_key_for( $idx );
        unless ( $key_sector ) {
            $self->reset;
            last;
        }

        $crumbs->[-1][1]++;
        $key = $key_sector->data;
        last;
    }

    return $key;
}

package DBM::Deep::Engine::Sector;

sub new {
    my $self = bless $_[1], $_[0];
    Scalar::Util::weaken( $self->{engine} );
    $self->_init;
    return $self;
}
sub _init {}
sub clone { die "Must be implemented in the child class" }

sub engine { $_[0]{engine} }
sub offset { $_[0]{offset} }
sub type   { $_[0]{type} }

sub free {
    my $self = shift;

    $self->engine->storage->print_at( $self->offset,
        $self->engine->SIG_FREE,
        chr(0) x ($self->size - 1),
    );

    $self->engine->_add_free_sector(
        $self->offset, $self->size,
    );

    return;
}

package DBM::Deep::Engine::Sector::Data;

our @ISA = qw( DBM::Deep::Engine::Sector );

# This is in bytes
sub size { return 256 }

sub clone {
    my $self = shift;
    return ref($self)->new({
        engine => $self->engine,
        data   => $self->data,
        type   => $self->type,
    });
}

package DBM::Deep::Engine::Sector::Scalar;

our @ISA = qw( DBM::Deep::Engine::Sector::Data );

sub free {
    my $self = shift;

    my $chain_loc = $self->chain_loc;

    $self->SUPER::free();

    if ( $chain_loc ) {
        $self->engine->_load_sector( $chain_loc )->free;
    }

    return;
}

sub type { $_[0]{engine}->SIG_DATA }
sub _init {
    my $self = shift;

    my $engine = $self->engine;

    unless ( $self->offset ) {
        my $data_section = $self->size - 3 - 1 * $engine->byte_size;

        $self->{offset} = $engine->_request_sector( $self->size );

        my $data = delete $self->{data};
        my $dlen = length $data;
        my $continue = 1;
        my $curr_offset = $self->offset;
        while ( $continue ) {

            my $next_offset = 0;

            my ($leftover, $this_len, $chunk);
            if ( $dlen > $data_section ) {
                $leftover = 0;
                $this_len = $data_section;
                $chunk = substr( $data, 0, $this_len );

                $dlen -= $data_section;
                $next_offset = $engine->_request_sector( $self->size );
                $data = substr( $data, $this_len );
            }
            else {
                $leftover = $data_section - $dlen;
                $this_len = $dlen;
                $chunk = $data;

                $continue = 0;
            }

            $engine->storage->print_at( $curr_offset,
                $self->type,                                     # Sector type
                pack( $StP{1}, 0 ),                              # Recycled counter
                pack( $StP{$engine->byte_size}, $next_offset ),  # Chain loc
                pack( $StP{1}, $this_len ),                      # Data length
                $chunk,                                          # Data to be stored in this sector
                chr(0) x $leftover,                              # Zero-fill the rest
            );

            $curr_offset = $next_offset;
        }

        return;
    }
}

sub data_length {
    my $self = shift;

    my $buffer = $self->engine->storage->read_at(
        $self->offset + 2 + $self->engine->byte_size, 1
    );

    return unpack( $StP{1}, $buffer );
}

sub chain_loc {
    my $self = shift;
    my $chain_loc = $self->engine->storage->read_at(
        $self->offset + 2, $self->engine->byte_size,
    );
    return unpack( $StP{$self->engine->byte_size}, $chain_loc );
}

sub data {
    my $self = shift;

    my $data;
    while ( 1 ) {
        my $chain_loc = $self->chain_loc;

        $data .= $self->engine->storage->read_at(
            $self->offset + 2 + $self->engine->byte_size + 1, $self->data_length,
        );

        last unless $chain_loc;

        $self = $self->engine->_load_sector( $chain_loc );
    }

    return $data;
}

package DBM::Deep::Engine::Sector::Null;

our @ISA = qw( DBM::Deep::Engine::Sector::Data );

sub type { $_[0]{engine}->SIG_NULL }
sub data_length { 0 }
sub data { return }

sub _init {
    my $self = shift;

    my $engine = $self->engine;

    unless ( $self->offset ) {
        my $leftover = $self->size - 3 - 1 * $engine->byte_size;

        $self->{offset} = $engine->_request_sector( $self->size );
        $engine->storage->print_at( $self->offset,
            $self->type,                          # Sector type
            pack( $StP{1}, 0 ),                   # Recycled counter
            pack( $StP{$engine->byte_size}, 0 ),  # Chain loc
            pack( $StP{1}, $self->data_length ),  # Data length
            chr(0) x $leftover,                   # Zero-fill the rest
        );

        return;
    }
}

package DBM::Deep::Engine::Sector::Reference;

our @ISA = qw( DBM::Deep::Engine::Sector::Data );

sub _init {
    my $self = shift;

    my $engine = $self->engine;

    unless ( $self->offset ) {
        my $classname = Scalar::Util::blessed( delete $self->{data} );
        my $leftover = $self->size - 4 - 2 * $engine->byte_size;

        my $class_offset = 0;
        if ( defined $classname ) {
            my $class_sector = DBM::Deep::Engine::Sector::Scalar->new({
                engine => $self->engine,
                data   => $classname,
            });
            $class_offset = $class_sector->offset;
        }

        $self->{offset} = $engine->_request_sector( $self->size );
        $engine->storage->print_at( $self->offset,
            $self->type,                                     # Sector type
            pack( $StP{1}, 0 ),                              # Recycled counter
            pack( $StP{$engine->byte_size}, 0 ),             # Index/BList loc
            pack( $StP{$engine->byte_size}, $class_offset ), # Classname loc
            chr(0) x $leftover,                              # Zero-fill the rest
        );

        return;
    }

    $self->{type} = $engine->storage->read_at( $self->offset, 1 );

    return;
}

sub get_data_for {
    my $self = shift;
    my ($args) = @_;

    # Assume that the head is not allowed unless otherwise specified.
    $args->{allow_head} = 0 unless exists $args->{allow_head};

    # Assume we don't create a new blist location unless otherwise specified.
    $args->{create} = 0 unless exists $args->{create};

    my $blist = $self->get_bucket_list({
        key_md5 => $args->{key_md5},
        create  => $args->{create},
    });
    return unless $blist && $blist->{found};

    # At this point, $blist knows where the md5 is. What it -doesn't- know yet
    # is whether or not this transaction has this key. That's part of the next
    # function call.
    my $location = $blist->get_data_location_for({
        allow_head => $args->{allow_head},
    }) or return;

    return $self->engine->_load_sector( $location );
}

sub write_data {
    my $self = shift;
    my ($args) = @_;

    my $blist = $self->get_bucket_list({
        key_md5 => $args->{key_md5},
        create  => 1,
    }) or die "How did write_data fail (no blist)?!\n";

    # Handle any transactional bookkeeping.
    if ( $self->engine->trans_id ) {
        if ( ! $blist->{found} ) {
            $blist->mark_deleted({
                trans_id => 0,
            });
        }
    }
    else {
        my @trans_ids = $self->engine->get_running_txn_ids;
        if ( $blist->{found} ) {
            if ( @trans_ids ) {
                my $old_value = $blist->get_data_for;
                foreach my $other_trans_id ( @trans_ids ) {
                    next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
                    #print "write_md5 to save a value\n";
                    $blist->write_md5({
                        trans_id => $other_trans_id,
                        key      => $args->{key},
                        key_md5  => $args->{key_md5},
                        value    => $old_value->clone,
                    });
                }
            }
        }
        else {
            if ( @trans_ids ) {
                foreach my $other_trans_id ( @trans_ids ) {
                    next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
                    $blist->mark_deleted({
                        trans_id => $other_trans_id,
                    });
                }
            }
        }
    }

    #XXX Is this safe to do transactionally?
    # Free the place we're about to write to.
    if ( $blist->get_data_location_for({ allow_head => 0 }) ) {
        $blist->get_data_for({ allow_head => 0 })->free;
    }

    $blist->write_md5({
        key      => $args->{key},
        key_md5  => $args->{key_md5},
        value    => $args->{value},
    });
}

sub delete_key {
    my $self = shift;
    my ($args) = @_;

    # XXX What should happen if this fails?
    my $blist = $self->get_bucket_list({
        key_md5 => $args->{key_md5},
    }) or die "How did delete_key fail (no blist)?!\n";

    # Save the location so that we can free the data
    my $location = $blist->get_data_location_for({
        allow_head => 0,
    });
    my $old_value = $location && $self->engine->_load_sector( $location );

    if ( $self->engine->trans_id == 0 ) {
        my @trans_ids = $self->engine->get_running_txn_ids;
        if ( @trans_ids ) {
            foreach my $other_trans_id ( @trans_ids ) {
                next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
                $blist->write_md5({
                    trans_id => $other_trans_id,
                    key      => $args->{key},
                    key_md5  => $args->{key_md5},
                    value    => $old_value->clone,
                });
            }
        }
    }

    $blist->mark_deleted( $args );

    my $data;
    if ( $old_value ) {
        $data = $old_value->data;
        $old_value->free;
    }

    return $data;
}

sub get_blist_loc {
    my $self = shift;

    my $e = $self->engine;
    my $blist_loc = $e->storage->read_at( $self->offset + 2, $e->byte_size );
    return unpack( $StP{$e->byte_size}, $blist_loc );
}

sub get_bucket_list {
    my $self = shift;
    my ($args) = @_;
    $args ||= {};

    # XXX Add in check here for recycling?

    my $engine = $self->engine;

    my $blist_loc = $self->get_blist_loc;

    # There's no index or blist yet
    unless ( $blist_loc ) {
        return unless $args->{create};

        my $blist = DBM::Deep::Engine::Sector::BucketList->new({
            engine  => $engine,
            key_md5 => $args->{key_md5},
        });

        $engine->storage->print_at( $self->offset + 2,
            pack( $StP{$engine->byte_size}, $blist->offset ),
        );

        return $blist;
    }

    return DBM::Deep::Engine::Sector::BucketList->new({
        engine  => $engine,
        offset  => $blist_loc,
        key_md5 => $args->{key_md5},
    });
}

sub get_classname {
    my $self = shift;

    my $class_offset = $self->engine->storage->read_at(
        $self->offset + 2 + 1 * $self->engine->byte_size, $self->engine->byte_size,
    );
    $class_offset = unpack ( $StP{$self->engine->byte_size}, $class_offset );

    return unless $class_offset;

    return $self->engine->_load_sector( $class_offset )->data;
}

sub data {
    my $self = shift;

    my $new_obj = DBM::Deep->new({
        type        => $self->type,
        base_offset => $self->offset,
        storage     => $self->engine->storage,
        engine      => $self->engine,
    });

    if ( $self->engine->storage->{autobless} ) {
        my $classname = $self->get_classname;
        if ( defined $classname ) {
            bless $new_obj, $classname;
        }
    }

    return $new_obj;
}

package DBM::Deep::Engine::Sector::BucketList;

our @ISA = qw( DBM::Deep::Engine::Sector );

sub _init {
    my $self = shift;

    my $engine = $self->engine;

    unless ( $self->offset ) {
        my $leftover = $self->size - $self->base_size;

        $self->{offset} = $engine->_request_sector( $self->size );
        $engine->storage->print_at( $self->offset,
            $engine->SIG_BLIST, # Sector type
            pack( $StP{1}, 0 ), # Recycled counter
            chr(0) x $leftover, # Zero-fill the data
        );
    }

    if ( $self->{key_md5} ) {
        $self->find_md5;
    }

    return $self;
}

sub base_size { 1 + 1 } # Sig + recycled counter

sub size {
    my $self = shift;
    unless ( $self->{size} ) {
        my $e = $self->engine;
        $self->{size} = $self->base_size + $e->max_buckets * $self->bucket_size; # Base + numbuckets * bucketsize
    }
    return $self->{size};
}

sub bucket_size {
    my $self = shift;
    unless ( $self->{bucket_size} ) {
        my $e = $self->engine;
        # Key + head (location) + transactions (location + staleness-counter)
        my $location_size = $e->byte_size + $e->num_txns * ( $e->byte_size + 4 );
        $self->{bucket_size} = $e->hash_size + $location_size;
    }
    return $self->{bucket_size};
}

sub has_md5 {
    my $self = shift;
    unless ( exists $self->{found} ) {
        $self->find_md5;
    }
    return $self->{found};
}

sub find_md5 {
    my $self = shift;

    $self->{found} = undef;
    $self->{idx}   = -1;

    # If we don't have an MD5, then what are we supposed to do?
    unless ( exists $self->{key_md5} ) {
        DBM::Deep->_throw_error( "Cannot find_md5 without a key_md5 set" );
    }

    my $e = $self->engine;
    foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
        my $potential = $e->storage->read_at(
            $self->offset + $self->base_size + $idx * $self->bucket_size, $e->hash_size,
        );

        if ( $potential eq $e->blank_md5 ) {
            $self->{idx} = $idx;
            return;
        }

        if ( $potential eq $self->{key_md5} ) {
            $self->{found} = 1;
            $self->{idx} = $idx;
            return;
        }
    }

    return;
}

sub write_md5 {
    my $self = shift;
    my ($args) = @_;

    DBM::Deep->_throw_error( "write_md5: no key" ) unless exists $args->{key};
    DBM::Deep->_throw_error( "write_md5: no key_md5" ) unless exists $args->{key_md5};
    DBM::Deep->_throw_error( "write_md5: no value" ) unless exists $args->{value};

    my $engine = $self->engine;

    $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};

    my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
    #print "Adding $args->{trans_id} -> $spot\n";
    $engine->add_entry( $args->{trans_id}, $spot );

    unless ($self->{found}) {
        my $key_sector = DBM::Deep::Engine::Sector::Scalar->new({
            engine => $engine,
            data   => $args->{key},
        });

        $engine->storage->print_at( $spot,
            $args->{key_md5},
            pack( $StP{$engine->byte_size}, $key_sector->offset ),
        );
    }

    my $loc = $spot
      + $engine->hash_size
      + $engine->byte_size
      + $args->{trans_id} * ( $engine->byte_size + 4 );

    $engine->storage->print_at( $loc,
        pack( $StP{$engine->byte_size}, $args->{value}->offset ),
        pack( 'N', $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
    );
}

sub mark_deleted {
    my $self = shift;
    my ($args) = @_;
    $args ||= {};

    my $engine = $self->engine;

    $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};

    my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
    $engine->add_entry( $args->{trans_id}, $spot );

    my $loc = $spot
      + $engine->hash_size
      + $engine->byte_size
      + $args->{trans_id} * ( $engine->byte_size + 4 );

    $engine->storage->print_at( $loc,
        pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
        pack( 'N', $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
    );
}

sub delete_md5 {
    my $self = shift;
    my ($args) = @_;

    my $engine = $self->engine;
    return undef unless $self->{found};

    # Save the location so that we can free the data
    my $location = $self->get_data_location_for({
        allow_head => 0,
    });
    my $key_sector = $self->get_key_for;

    #XXX This isn't going to work right and you know it! This eradicates data
    # that we're not ready to eradicate just yet.
    my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
    $engine->storage->print_at( $spot,
        $engine->storage->read_at(
            $spot + $self->bucket_size,
            $self->bucket_size * ( $engine->num_txns - $self->{idx} - 1 ),
        ),
        chr(0) x $self->bucket_size,
    );

    $key_sector->free;

    my $data_sector = $self->engine->_load_sector( $location );
    my $data = $data_sector->data;
    $data_sector->free;

    return $data;
}

sub get_data_location_for {
    my $self = shift;
    my ($args) = @_;
    $args ||= {};

    $args->{allow_head} = 0 unless exists $args->{allow_head};
    $args->{trans_id}   = $self->engine->trans_id unless exists $args->{trans_id};
    $args->{idx}        = $self->{idx} unless exists $args->{idx};

    my $e = $self->engine;

    my $spot = $self->offset + $self->base_size
      + $args->{idx} * $self->bucket_size
      + $e->hash_size
      + $e->byte_size
      + $args->{trans_id} * ( $e->byte_size + 4 );

    my $buffer = $e->storage->read_at(
        $spot,
        $e->byte_size + 4,
    );
    my ($loc, $staleness) = unpack( $StP{$e->byte_size} . ' N', $buffer );

    # We have found an entry that is old, so get rid of it
    if ( $staleness != $e->get_txn_staleness_counter( $args->{trans_id} ) ) {
        $e->storage->print_at(
            $spot,
            pack( $StP{$e->byte_size} . ' N', (0) x 2 ), 
        );
        $loc = 0;
    }

    # If we're in a transaction and we never wrote to this location, try the
    # HEAD instead.
    if ( $args->{trans_id} && !$loc && $args->{allow_head} ) {
        return $self->get_data_location_for({
            trans_id   => 0,
            allow_head => 1,
            idx        => $args->{idx},
        });
    }
    return $loc <= 1 ? 0 : $loc;
}

sub get_data_for {
    my $self = shift;
    my ($args) = @_;
    $args ||= {};

    return unless $self->{found};
    my $location = $self->get_data_location_for({
        allow_head => $args->{allow_head},
    });
    return $self->engine->_load_sector( $location );
}

sub get_key_for {
    my $self = shift;
    my ($idx) = @_;
    $idx = $self->{idx} unless defined $idx;

    my $location = $self->engine->storage->read_at(
        $self->offset + $self->base_size + $idx * $self->bucket_size + $self->engine->hash_size,
        $self->engine->byte_size,
    );
    $location = unpack( $StP{$self->engine->byte_size}, $location );
    return unless $location;
    return $self->engine->_load_sector( $location );
}

1;
__END__