[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

        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 );

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

    my $value_sector = $blist->get_data_for( $key_md5 );
    if ( !$value_sector ) {
        # Autovivification
        $value_sector = DBM::Deep::Engine::Sector::Null->new({
            engine => $self,
            data   => undef,
        });

        $blist->write_md5( $key_md5, $key, $value_sector->offset );
    }

    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 $key_md5 = $self->_apply_digest( $key );

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

    # exists() returns 1 or '' for true/false.
    return $blist->has_md5( $key_md5 ) ? 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";

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

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

    return $blist->delete_md5( $key_md5 );
}

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

    # 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";

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

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

    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';
    }

    if ( $blist->has_md5( $key_md5 ) ) {
        $blist->get_data_for( $key_md5 )->free;
    }

    my $value_sector = $class->new({
        engine => $self,
        data   => $value,
        type   => $type,
    });

    $blist->write_md5( $key_md5, $key, $value_sector->offset );

    # 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;
}

# begin_work
sub begin_transaction {
    my $self = shift;
}

# rollback
sub end_transaction {
    my $self = shift;
}

# commit
sub commit_transaction {
    my $self = shift;
}

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

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

    sub _write_file_header {
        my $self = shift;

        my $header_var = 1 + 1 + 4 + 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 ),                   # Running transactions
            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 + 6 );

        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 + 6 );

        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 + 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,
        });
    }
    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_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},
        trans_id    => $args->{trans_id},
    }, $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;
        }

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

        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 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 }

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;

        my $data = delete $self->{data};

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

        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_blist_loc {
    my $self = shift;

    my $engine = $self->engine;
    my $blist_loc = $engine->storage->read_at( $self->offset + 2, $engine->byte_size );
    return unpack( $StP{$engine->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,
        });
        $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,
    });
}

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 idx_for_txn { return $_[1] + 1 }

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
        );
    }

    return $self;
}

sub base_size { 2 } # Sig + recycled counter

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

sub bucket_size {
    my $self = shift;
    my $e = $self->engine;
    # Key + transactions
    my $locs_size = (1 + $e->num_txns ) * $e->byte_size;
    return $e->hash_size + $locs_size;
}

sub has_md5 {
    my $self = shift;
    my ($found, $idx) = $self->find_md5( @_ );
    return $found;
}

sub find_md5 {
    my $self = shift;
    my ($md5) = @_;

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

        return (undef, $idx) if $potential eq $self->engine->blank_md5;
        return (1, $idx) if $md5 eq $potential;
    }

    return;
}

sub write_md5 {
    my $self = shift;
    my ($md5, $key, $value_loc) = @_;

    my $engine = $self->engine;
    my ($found, $idx) = $self->find_md5( $md5 );
    my $spot = $self->offset + $self->base_size + $idx * $self->bucket_size;

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

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

    $engine->storage->print_at( $spot + $self->engine->hash_size + $self->engine->byte_size,
        pack( $StP{$engine->byte_size}, $value_loc ), # The pointer to the data in the HEAD
    );
}

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

    my $engine = $self->engine;
    my ($found, $idx) = $self->find_md5( $md5 );
    return undef unless $found;

    # Save the location so that we can free the data
    my $location = $self->get_data_location_for( $idx );
    my $key_sector = $self->get_key_for( $idx );

    my $spot = $self->offset + $self->base_size + $idx * $self->bucket_size;
    $engine->storage->print_at( $spot,
        $engine->storage->read_at(
            $spot + $self->bucket_size,
            $self->bucket_size * ( $engine->num_txns - $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 ($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,
        $self->engine->byte_size,
    );
    return unpack( $StP{$self->engine->byte_size}, $location );
}

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

    my ($found, $idx) = $self->find_md5( $md5 );
    return unless $found;
    my $location = $self->get_data_location_for( $idx );
    return $self->engine->_load_sector( $location );
}

sub get_key_for {
    my $self = shift;
    my ($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__