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

package DBM::Deep::Engine2;

use base 'DBM::Deep::Engine';

use 5.6.0;

use strict;
use warnings;

our $VERSION = q(0.99_03);

use Fcntl qw( :DEFAULT :flock );
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     }

# This is the transaction ID for the HEAD
sub HEAD () { 0 }

sub read_value {
    my $self = shift;
    my ($trans_id, $base_offset, $key) = @_;
    
    my ($_val_offset, $_is_del) = $self->_find_value_offset({
        offset     => $base_offset,
        trans_id   => $trans_id,
        allow_head => 1,
    });
    die "Attempt to use a deleted value" if $_is_del;
    die "Internal error!" if !$_val_offset;

    my ($key_tag) = $self->_find_key_offset({
        offset  => $_val_offset,
        key_md5 => $self->_apply_digest( $key ),
    });
    return if !$key_tag;

    my ($val_offset, $is_del) = $self->_find_value_offset({
        offset     => $key_tag->{start},
        trans_id   => $trans_id,
        allow_head => 1,
    });
    return if $is_del;
    die "Internal error!" if !$val_offset;

    return $self->_read_value({
        keyloc => $key_tag->{start},
        offset => $val_offset,
        key    => $key,
    });
}

sub key_exists {
    my $self = shift;
    my ($trans_id, $base_offset, $key) = @_;
    
    my ($_val_offset, $_is_del) = $self->_find_value_offset({
        offset     => $base_offset,
        trans_id   => $trans_id,
        allow_head => 1,
    });
    die "Attempt to use a deleted value" if $_is_del;
    die "Internal error!" if !$_val_offset;

    my ($key_tag) = $self->_find_key_offset({
        offset  => $_val_offset,
        key_md5 => $self->_apply_digest( $key ),
    });
    return '' if !$key_tag->{start};

    my ($val_offset, $is_del) = $self->_find_value_offset({
        offset     => $key_tag->{start},
        trans_id   => $trans_id,
        allow_head => 1,
    });
    die "Internal error!" if !$_val_offset;

    return '' if $is_del;

    return 1;
}

sub get_next_key {
    my $self = shift;
    my ($trans_id, $base_offset) = @_;

    my ($_val_offset, $_is_del) = $self->_find_value_offset({
        offset     => $base_offset,
        trans_id   => $trans_id,
        allow_head => 1,
    });
    die "Attempt to use a deleted value" if $_is_del;
    die "Internal error!" if !$_val_offset;

    # If the previous key was not specifed, start at the top and
    # return the first one found.
    my $temp;
    if ( @_ > 2 ) {
        $temp = {
            prev_md5    => $self->_apply_digest($_[2]),
            return_next => 0,
        };
    }
    else {
        $temp = {
            prev_md5    => chr(0) x $self->{hash_size},
            return_next => 1,
        };
    }

    local $::DEBUG = 1;
    print "get_next_key: $_val_offset\n" if $::DEBUG;
    return $self->traverse_index( $temp, $_val_offset, 0 );
}

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

    my ($_val_offset, $_is_del) = $self->_find_value_offset({
        offset     => $base_offset,
        trans_id   => $trans_id,
        allow_head => 1,
    });
    die "Attempt to use a deleted value" if $_is_del;
    die "Internal error!" if !$_val_offset;

    my ($key_tag, $bucket_tag) = $self->_find_key_offset({
        offset  => $_val_offset,
        key_md5 => $self->_apply_digest( $key ),
    });
    return if !$key_tag->{start};

    my $value = $self->read_value( $trans_id, $base_offset, $key );
    my $value = $self->read_value( $trans_id, $base_offset, $key );
    if ( $trans_id ) {
        $self->_mark_as_deleted({
            tag      => $key_tag,
            trans_id => $trans_id,
        });
    }
    else {
        if ( my @transactions = $self->_storage->current_transactions ) {
            foreach my $other_trans_id ( @transactions ) {
                next if $self->_has_keyloc_entry({
                    tag      => $key_tag,
                    trans_id => $other_trans_id,
                });
                $self->write_value( $other_trans_id, $base_offset, $key, $value );
            }
        }

        $self->_mark_as_deleted({
            tag      => $key_tag,
            trans_id => $trans_id,
        });
#        $self->_remove_key_offset({
#            offset  => $_val_offset,
#            key_md5 => $self->_apply_digest( $key ),
#        });
    }

    return $value;
}

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

    # This verifies that only supported values will be stored.
    {
        my $r = Scalar::Util::reftype( $value );

        last if !defined $r;
        last if $r eq 'HASH';
        last if $r eq 'ARRAY';

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

    my ($_val_offset, $_is_del) = $self->_find_value_offset({
        offset     => $base_offset,
        trans_id   => $trans_id,
        allow_head => 1,
    });
    die "Attempt to use a deleted value" if $_is_del;
    die "Internal error!" if !$_val_offset;

    my ($key_tag, $bucket_tag) = $self->_find_key_offset({
        offset  => $_val_offset,
        key_md5 => $self->_apply_digest( $key ),
        create  => 1,
    });
    die "Cannot find/create new key offset!" if !$key_tag->{start};

    if ( $trans_id ) {
        if ( $key_tag->{is_new} ) {
            # Must mark the HEAD as deleted because it doesn't exist
            $self->_mark_as_deleted({
                tag      => $key_tag,
                trans_id => HEAD,
            });
        }
    }
    else {
        # If the HEAD isn't new, then we must take other transactions
        # into account. If it is, then there can be no other transactions.
        if ( !$key_tag->{is_new} ) {
            my $old_value = $self->read_value( $trans_id, $base_offset, $key );
            if ( my @transactions = $self->_storage->current_transactions ) {
                foreach my $other_trans_id ( @transactions ) {
                    next if $self->_has_keyloc_entry({
                        tag      => $key_tag,
                        trans_id => $other_trans_id,
                    });
                    $self->write_value( $other_trans_id, $base_offset, $key, $old_value );
                }
            }
        }
    }

    my $value_loc = $self->_storage->request_space( 
        $self->_length_needed( $value, $key ),
    );

    $self->_add_key_offset({
        tag      => $key_tag,
        trans_id => $trans_id,
        loc      => $value_loc,
    });

    $self->_write_value( $key_tag->{start}, $value_loc, $key, $value, $key );

    return 1;
}

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

    my $key_tag = $self->load_tag( $args->{offset} );

    my @head;
    for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
        my ($loc, $trans_id, $is_deleted) = unpack(
            "$self->{long_pack} C C",
            substr( $key_tag->{content}, $i * $self->{key_size}, $self->{key_size} ),
        );

        if ( $trans_id == HEAD ) {
            @head = ($loc, $is_deleted);
        }

        next if $loc && $args->{trans_id} != $trans_id;
        return( $loc, $is_deleted );
    }

    return @head if $args->{allow_head};
    return;
}

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

    my $bucket_tag = $self->load_tag( $args->{offset} )
        or $self->_throw_error( "INTERNAL ERROR - Cannot find tag" );

    #XXX What happens when $ch >= $self->{hash_size} ??
    for (my $ch = 0; $bucket_tag->{signature} ne SIG_BLIST; $ch++) {
        my $num = ord substr($args->{key_md5}, $ch, 1);

        my $ref_loc = $bucket_tag->{offset} + ($num * $self->{long_size});
        $bucket_tag = $self->index_lookup( $bucket_tag, $num );

        if (!$bucket_tag) {
            return if !$args->{create};

            my $loc = $self->_storage->request_space(
                $self->tag_size( $self->{bucket_list_size} ),
            );

            $self->_storage->print_at( $ref_loc, pack($self->{long_pack}, $loc) );

            $bucket_tag = $self->write_tag(
                $loc, SIG_BLIST,
                chr(0)x$self->{bucket_list_size},
            );

            $bucket_tag->{ref_loc} = $ref_loc;
            $bucket_tag->{ch} = $ch;
            $bucket_tag->{is_new} = 1;

            last;
        }

        $bucket_tag->{ch} = $ch;
        $bucket_tag->{ref_loc} = $ref_loc;
    }

    # Need to create a new keytag, too
    if ( $bucket_tag->{is_new} ) {
        my $keytag_loc = $self->_storage->request_space(
            $self->tag_size( $self->{keyloc_size} ),
        );

        substr( $bucket_tag->{content}, 0, $self->{key_size} ) =
            $args->{key_md5} . pack( "$self->{long_pack}", $keytag_loc );

        $self->_storage->print_at( $bucket_tag->{offset}, $bucket_tag->{content} );

        my $key_tag = $self->write_tag(
            $keytag_loc, SIG_KEYS,
            chr(0)x$self->{keyloc_size},
        );

        return( $key_tag, $bucket_tag );
    }
    else {
        my ($key, $subloc, $index);
        BUCKET:
        for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
            ($key, $subloc) = $self->_get_key_subloc(
                $bucket_tag->{content}, $i,
            );

            next BUCKET if $subloc && $key ne $args->{key_md5};

            # Keep track of where we are, in case we need to create a new
            # entry.
            $index = $i;
            last;
        }

        # If we have a subloc to return or we don't want to create a new
        # entry, we need to return now.
        $args->{create} ||= 0;
        return ($self->load_tag( $subloc ), $bucket_tag) if $subloc || !$args->{create};

        my $keytag_loc = $self->_storage->request_space(
            $self->tag_size( $self->{keyloc_size} ),
        );

        # There's space left in this bucket
        if ( defined $index ) {
            substr( $bucket_tag->{content}, $index * $self->{key_size}, $self->{key_size} ) =
                $args->{key_md5} . pack( "$self->{long_pack}", $keytag_loc );

            $self->_storage->print_at( $bucket_tag->{offset}, $bucket_tag->{content} );
        }
        # We need to split the index
        else {
            $self->split_index( $bucket_tag, $args->{key_md5}, $keytag_loc );
        }

        my $key_tag = $self->write_tag(
            $keytag_loc, SIG_KEYS,
            chr(0)x$self->{keyloc_size},
        );

        return( $key_tag, $bucket_tag );
    }

    return;
}

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

    return $self->read_from_loc( $args->{keyloc}, $args->{offset}, $args->{key} );
}

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

    my $is_changed;
    for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
        my ($loc, $trans_id, $is_deleted) = unpack(
            "$self->{long_pack} C C",
            substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ),
        );

        last unless $loc || $is_deleted;

        if ( $trans_id == $args->{trans_id} ) {
            substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ) = pack(
                "$self->{long_pack} C C",
                $loc, $trans_id, 1,
            );
            $is_changed = 1;
            last;
        }
    }

    if ( $is_changed ) {
        $self->_storage->print_at(
            $args->{tag}{offset}, $args->{tag}{content},
        );
    }

    return 1;
}

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

    for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
        my ($loc, $trans_id, $is_deleted) = unpack(
            "$self->{long_pack} C C",
            substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ),
        );

        return 1 if $trans_id == $args->{trans_id};
    }

    return;
}

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

    my $is_changed;
    for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
        my ($loc, $trans_id, $is_deleted) = unpack(
            "$self->{long_pack} C C",
            substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ),
        );

        if ( $trans_id == $args->{trans_id} ) {
            substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ) = '';
            $args->{tag}{content} .= chr(0) x $self->{key_size};
            $is_changed = 1;
            redo;
        }
    }

    if ( $is_changed ) {
        $self->_storage->print_at(
            $args->{tag}{offset}, $args->{tag}{content},
        );
    }

    return 1;
}

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

    my $is_changed;
    for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
        my ($loc, $trans_id, $is_deleted) = unpack(
            "$self->{long_pack} C C",
            substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ),
        );

        if ( $trans_id == $args->{trans_id} || (!$loc && !$is_deleted) ) {
            substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ) = pack(
                "$self->{long_pack} C C",
                $args->{loc}, $args->{trans_id}, 0,
            );
            $is_changed = 1;
            last;
        }
    }

    if ( $is_changed ) {
        $self->_storage->print_at(
            $args->{tag}{offset}, $args->{tag}{content},
        );
    }
    else {
        die "Why didn't _add_key_offset() change something?!\n";
    }

    return 1;
}

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

    # Need to remove use of $fh here
    my $fh = $self->_storage->{fh};
    flock $fh, LOCK_EX;

    #XXX The duplication of calculate_sizes needs to go away
    unless ( $obj->{base_offset} ) {
        my $bytes_read = $self->read_file_header;

        $self->calculate_sizes;

        ##
        # File is empty -- write header and master index
        ##
        if (!$bytes_read) {
            $self->_storage->audit( "# Database created on" );

            $self->write_file_header;

            $obj->{base_offset} = $self->_storage->request_space(
                $self->tag_size( $self->{keyloc_size} ),
            );

            my $value_spot = $self->_storage->request_space(
                $self->tag_size( $self->{index_size} ),
            );

            $self->write_tag(
                $obj->{base_offset}, SIG_KEYS,
                pack( "$self->{long_pack} C C", $value_spot, HEAD, 0 ),
                chr(0) x ($self->{index_size} - $self->{key_size}),
            );

            $self->write_tag(
                $value_spot, $obj->_type,
                chr(0)x$self->{index_size},
            );

            # Flush the filehandle
            my $old_fh = select $fh;
            my $old_af = $|; $| = 1; $| = $old_af;
            select $old_fh;
        }
        else {
            $obj->{base_offset} = $bytes_read;

            my ($_val_offset, $_is_del) = $self->_find_value_offset({
                offset     => $obj->{base_offset},
                trans_id   => HEAD,
                allow_head => 1,
            });
            die "Attempt to use a deleted value" if $_is_del;
            die "Internal error!" if !$_val_offset;

            ##
            # Get our type from master index header
            ##
            my $tag = $self->load_tag($_val_offset);
            unless ( $tag ) {
                flock $fh, LOCK_UN;
                $self->_throw_error("Corrupted file, no master index record");
            }

            unless ($obj->_type eq $tag->{signature}) {
                flock $fh, LOCK_UN;
                $self->_throw_error("File type mismatch");
            }
        }
    }
    else {
        $self->calculate_sizes;
    }

    #XXX We have to make sure we don't mess up when autoflush isn't turned on
    $self->_storage->set_inode;

    flock $fh, LOCK_UN;

    return 1;
}

1;
__END__
Commit	Line	Data
fb451ba6	1	package DBM::Deep::Engine2;
	2
	3	use base 'DBM::Deep::Engine';
	4
	5	use 5.6.0;
	6
	7	use strict;
	8	use warnings;
	9
	10	our $VERSION = q(0.99_03);
	11
	12	use Fcntl qw( :DEFAULT :flock );
	13	use Scalar::Util ();
	14
	15	# File-wide notes:
	16	# * Every method in here assumes that the _storage has been appropriately
	17	# safeguarded. This can be anything from flock() to some sort of manual
	18	# mutex. But, it's the caller's responsability to make sure that this has
	19	# been done.
	20
	21	# Setup file and tag signatures. These should never change.
	22	sub SIG_FILE () { 'DPDB' }
	23	sub SIG_HEADER () { 'h' }
	24	sub SIG_INTERNAL () { 'i' }
	25	sub SIG_HASH () { 'H' }
	26	sub SIG_ARRAY () { 'A' }
	27	sub SIG_NULL () { 'N' }
	28	sub SIG_DATA () { 'D' }
	29	sub SIG_INDEX () { 'I' }
	30	sub SIG_BLIST () { 'B' }
	31	sub SIG_FREE () { 'F' }
	32	sub SIG_KEYS () { 'K' }
	33	sub SIG_SIZE () { 1 }
	34
	35	# This is the transaction ID for the HEAD
	36	sub HEAD () { 0 }
	37
	38	sub read_value {
	39	my $self = shift;
	40	my ($trans_id, $base_offset, $key) = @_;
	41
	42	my ($_val_offset, $_is_del) = $self->_find_value_offset({
	43	offset => $base_offset,
	44	trans_id => $trans_id,
	45	allow_head => 1,
	46	});
	47	die "Attempt to use a deleted value" if $_is_del;
	48	die "Internal error!" if !$_val_offset;
	49
f5677eab	50	my ($key_tag) = $self->_find_key_offset({
fb451ba6	51	offset => $_val_offset,
	52	key_md5 => $self->_apply_digest( $key ),
	53	});
f5677eab	54	return if !$key_tag;
fb451ba6	55
fb451ba6	56	my ($val_offset, $is_del) = $self->_find_value_offset({
f5677eab	57	offset => $key_tag->{start},
fb451ba6	58	trans_id => $trans_id,
	59	allow_head => 1,
	60	});
	61	return if $is_del;
	62	die "Internal error!" if !$val_offset;
	63
	64	return $self->_read_value({
f5677eab	65	keyloc => $key_tag->{start},
fb451ba6	66	offset => $val_offset,
1ad1fc2b	67	key => $key,
fb451ba6	68	});
	69	}
	70
	71	sub key_exists {
	72	my $self = shift;
	73	my ($trans_id, $base_offset, $key) = @_;
	74
	75	my ($_val_offset, $_is_del) = $self->_find_value_offset({
	76	offset => $base_offset,
	77	trans_id => $trans_id,
	78	allow_head => 1,
	79	});
	80	die "Attempt to use a deleted value" if $_is_del;
	81	die "Internal error!" if !$_val_offset;
	82
f5677eab	83	my ($key_tag) = $self->_find_key_offset({
fb451ba6	84	offset => $_val_offset,
	85	key_md5 => $self->_apply_digest( $key ),
	86	});
f5677eab	87	return '' if !$key_tag->{start};
fb451ba6	88
fb451ba6	89	my ($val_offset, $is_del) = $self->_find_value_offset({
f5677eab	90	offset => $key_tag->{start},
fb451ba6	91	trans_id => $trans_id,
	92	allow_head => 1,
	93	});
f5677eab	94	die "Internal error!" if !$_val_offset;
fb451ba6	95
129ea236	96	return '' if $is_del;
fb451ba6	97
129ea236	98	return 1;
fb451ba6	99	}
	100
	101	sub get_next_key {
	102	my $self = shift;
129ea236	103	my ($trans_id, $base_offset) = @_;
	104
	105	my ($_val_offset, $_is_del) = $self->_find_value_offset({
	106	offset => $base_offset,
	107	trans_id => $trans_id,
	108	allow_head => 1,
	109	});
	110	die "Attempt to use a deleted value" if $_is_del;
	111	die "Internal error!" if !$_val_offset;
fb451ba6	112
	113	# If the previous key was not specifed, start at the top and
	114	# return the first one found.
	115	my $temp;
129ea236	116	if ( @_ > 2 ) {
fb451ba6	117	$temp = {
129ea236	118	prev_md5 => $self->_apply_digest($_[2]),
fb451ba6	119	return_next => 0,
	120	};
	121	}
	122	else {
	123	$temp = {
	124	prev_md5 => chr(0) x $self->{hash_size},
	125	return_next => 1,
	126	};
	127	}
	128
1ad1fc2b	129	local $::DEBUG = 1;
1ad1fc2b	130	print "get_next_key: $_val_offset\n" if $::DEBUG;
129ea236	131	return $self->traverse_index( $temp, $_val_offset, 0 );
fb451ba6	132	}
	133
	134	sub delete_key {
	135	my $self = shift;
	136	my ($trans_id, $base_offset, $key) = @_;
	137
	138	my ($_val_offset, $_is_del) = $self->_find_value_offset({
	139	offset => $base_offset,
	140	trans_id => $trans_id,
	141	allow_head => 1,
	142	});
	143	die "Attempt to use a deleted value" if $_is_del;
	144	die "Internal error!" if !$_val_offset;
	145
f5677eab	146	my ($key_tag, $bucket_tag) = $self->_find_key_offset({
fb451ba6	147	offset => $_val_offset,
	148	key_md5 => $self->_apply_digest( $key ),
	149	});
f5677eab	150	return if !$key_tag->{start};
fb451ba6	151
129ea236	152	my $value = $self->read_value( $trans_id, $base_offset, $key );
11eb1592	153	my $value = $self->read_value( $trans_id, $base_offset, $key );
fb451ba6	154	if ( $trans_id ) {
	155	$self->_mark_as_deleted({
	156	tag => $key_tag,
	157	trans_id => $trans_id,
	158	});
	159	}
	160	else {
fb451ba6	161	if ( my @transactions = $self->_storage->current_transactions ) {
	162	foreach my $other_trans_id ( @transactions ) {
	163	next if $self->_has_keyloc_entry({
	164	tag => $key_tag,
	165	trans_id => $other_trans_id,
	166	});
	167	$self->write_value( $other_trans_id, $base_offset, $key, $value );
	168	}
	169	}
129ea236	170
	171	$self->_mark_as_deleted({
	172	tag => $key_tag,
	173	trans_id => $trans_id,
	174	});
	175	# $self->_remove_key_offset({
	176	# offset => $_val_offset,
	177	# key_md5 => $self->_apply_digest( $key ),
	178	# });
fb451ba6	179	}
fb451ba6	180
129ea236	181	return $value;
fb451ba6	182	}
	183
	184	sub write_value {
	185	my $self = shift;
	186	my ($trans_id, $base_offset, $key, $value) = @_;
	187
	188	# This verifies that only supported values will be stored.
	189	{
	190	my $r = Scalar::Util::reftype( $value );
	191
	192	last if !defined $r;
	193	last if $r eq 'HASH';
	194	last if $r eq 'ARRAY';
	195
	196	$self->_throw_error(
	197	"Storage of references of type '$r' is not supported."
	198	);
	199	}
	200
	201	my ($_val_offset, $_is_del) = $self->_find_value_offset({
	202	offset => $base_offset,
	203	trans_id => $trans_id,
	204	allow_head => 1,
	205	});
	206	die "Attempt to use a deleted value" if $_is_del;
	207	die "Internal error!" if !$_val_offset;
	208
f5677eab	209	my ($key_tag, $bucket_tag) = $self->_find_key_offset({
fb451ba6	210	offset => $_val_offset,
	211	key_md5 => $self->_apply_digest( $key ),
	212	create => 1,
	213	});
f5677eab	214	die "Cannot find/create new key offset!" if !$key_tag->{start};
fb451ba6	215
fb451ba6	216	if ( $trans_id ) {
f5677eab	217	if ( $key_tag->{is_new} ) {
fb451ba6	218	# Must mark the HEAD as deleted because it doesn't exist
	219	$self->_mark_as_deleted({
	220	tag => $key_tag,
	221	trans_id => HEAD,
	222	});
	223	}
	224	}
	225	else {
	226	# If the HEAD isn't new, then we must take other transactions
	227	# into account. If it is, then there can be no other transactions.
f5677eab	228	if ( !$key_tag->{is_new} ) {
fb451ba6	229	my $old_value = $self->read_value( $trans_id, $base_offset, $key );
	230	if ( my @transactions = $self->_storage->current_transactions ) {
	231	foreach my $other_trans_id ( @transactions ) {
	232	next if $self->_has_keyloc_entry({
	233	tag => $key_tag,
	234	trans_id => $other_trans_id,
	235	});
	236	$self->write_value( $other_trans_id, $base_offset, $key, $old_value );
	237	}
	238	}
	239	}
	240	}
	241
40956c06	242	my $value_loc = $self->_storage->request_space(
	243	$self->_length_needed( $value, $key ),
	244	);
	245
	246	$self->_add_key_offset({
	247	tag => $key_tag,
	248	trans_id => $trans_id,
	249	loc => $value_loc,
fb451ba6	250	});
fb451ba6	251
f5677eab	252	$self->_write_value( $key_tag->{start}, $value_loc, $key, $value, $key );
40956c06	253
fb451ba6	254	return 1;
	255	}
	256
	257	sub _find_value_offset {
	258	my $self = shift;
	259	my ($args) = @_;
	260
	261	my $key_tag = $self->load_tag( $args->{offset} );
	262
	263	my @head;
	264	for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
	265	my ($loc, $trans_id, $is_deleted) = unpack(
	266	"$self->{long_pack} C C",
	267	substr( $key_tag->{content}, $i * $self->{key_size}, $self->{key_size} ),
	268	);
	269
	270	if ( $trans_id == HEAD ) {
	271	@head = ($loc, $is_deleted);
	272	}
	273
	274	next if $loc && $args->{trans_id} != $trans_id;
	275	return( $loc, $is_deleted );
	276	}
	277
	278	return @head if $args->{allow_head};
	279	return;
	280	}
	281
fb451ba6	282	sub _find_key_offset {
	283	my $self = shift;
	284	my ($args) = @_;
	285
	286	my $bucket_tag = $self->load_tag( $args->{offset} )
	287	or $self->_throw_error( "INTERNAL ERROR - Cannot find tag" );
	288
fb451ba6	289	#XXX What happens when $ch >= $self->{hash_size} ??
	290	for (my $ch = 0; $bucket_tag->{signature} ne SIG_BLIST; $ch++) {
	291	my $num = ord substr($args->{key_md5}, $ch, 1);
	292
	293	my $ref_loc = $bucket_tag->{offset} + ($num * $self->{long_size});
	294	$bucket_tag = $self->index_lookup( $bucket_tag, $num );
	295
	296	if (!$bucket_tag) {
	297	return if !$args->{create};
	298
	299	my $loc = $self->_storage->request_space(
	300	$self->tag_size( $self->{bucket_list_size} ),
	301	);
	302
	303	$self->_storage->print_at( $ref_loc, pack($self->{long_pack}, $loc) );
	304
	305	$bucket_tag = $self->write_tag(
	306	$loc, SIG_BLIST,
	307	chr(0)x$self->{bucket_list_size},
	308	);
	309
	310	$bucket_tag->{ref_loc} = $ref_loc;
	311	$bucket_tag->{ch} = $ch;
	312	$bucket_tag->{is_new} = 1;
	313
	314	last;
	315	}
	316
	317	$bucket_tag->{ch} = $ch;
	318	$bucket_tag->{ref_loc} = $ref_loc;
	319	}
	320
	321	# Need to create a new keytag, too
	322	if ( $bucket_tag->{is_new} ) {
	323	my $keytag_loc = $self->_storage->request_space(
	324	$self->tag_size( $self->{keyloc_size} ),
	325	);
	326
	327	substr( $bucket_tag->{content}, 0, $self->{key_size} ) =
	328	$args->{key_md5} . pack( "$self->{long_pack}", $keytag_loc );
	329
	330	$self->_storage->print_at( $bucket_tag->{offset}, $bucket_tag->{content} );
	331
f5677eab	332	my $key_tag = $self->write_tag(
fb451ba6	333	$keytag_loc, SIG_KEYS,
	334	chr(0)x$self->{keyloc_size},
	335	);
	336
f5677eab	337	return( $key_tag, $bucket_tag );
fb451ba6	338	}
fb451ba6	339	else {
40956c06	340	my ($key, $subloc, $index);
fb451ba6	341	BUCKET:
fb451ba6	342	for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
40956c06	343	($key, $subloc) = $self->_get_key_subloc(
fb451ba6	344	$bucket_tag->{content}, $i,
	345	);
	346
	347	next BUCKET if $subloc && $key ne $args->{key_md5};
40956c06	348
	349	# Keep track of where we are, in case we need to create a new
	350	# entry.
	351	$index = $i;
	352	last;
fb451ba6	353	}
40956c06	354
f5677eab	355	# If we have a subloc to return or we don't want to create a new
	356	# entry, we need to return now.
	357	$args->{create} \|\|= 0;
f5677eab	358	return ($self->load_tag( $subloc ), $bucket_tag) if $subloc \|\| !$args->{create};
40956c06	359
	360	my $keytag_loc = $self->_storage->request_space(
	361	$self->tag_size( $self->{keyloc_size} ),
	362	);
	363
	364	# There's space left in this bucket
	365	if ( defined $index ) {
	366	substr( $bucket_tag->{content}, $index * $self->{key_size}, $self->{key_size} ) =
	367	$args->{key_md5} . pack( "$self->{long_pack}", $keytag_loc );
	368
	369	$self->_storage->print_at( $bucket_tag->{offset}, $bucket_tag->{content} );
	370	}
	371	# We need to split the index
	372	else {
	373	$self->split_index( $bucket_tag, $args->{key_md5}, $keytag_loc );
	374	}
	375
f5677eab	376	my $key_tag = $self->write_tag(
40956c06	377	$keytag_loc, SIG_KEYS,
	378	chr(0)x$self->{keyloc_size},
	379	);
	380
f5677eab	381	return( $key_tag, $bucket_tag );
fb451ba6	382	}
	383
	384	return;
	385	}
	386
	387	sub _read_value {
	388	my $self = shift;
	389	my ($args) = @_;
	390
1ad1fc2b	391	return $self->read_from_loc( $args->{keyloc}, $args->{offset}, $args->{key} );
fb451ba6	392	}
	393
	394	sub _mark_as_deleted {
	395	my $self = shift;
	396	my ($args) = @_;
	397
	398	my $is_changed;
	399	for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
	400	my ($loc, $trans_id, $is_deleted) = unpack(
	401	"$self->{long_pack} C C",
	402	substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ),
	403	);
	404
129ea236	405	last unless $loc \|\| $is_deleted;
fb451ba6	406
	407	if ( $trans_id == $args->{trans_id} ) {
	408	substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ) = pack(
	409	"$self->{long_pack} C C",
	410	$loc, $trans_id, 1,
40956c06	411	);
129ea236	412	$is_changed = 1;
129ea236	413	last;
fb451ba6	414	}
	415	}
	416
	417	if ( $is_changed ) {
	418	$self->_storage->print_at(
	419	$args->{tag}{offset}, $args->{tag}{content},
	420	);
	421	}
	422
	423	return 1;
	424	}
	425
	426	sub _has_keyloc_entry {
	427	my $self = shift;
	428	my ($args) = @_;
	429
	430	for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
	431	my ($loc, $trans_id, $is_deleted) = unpack(
	432	"$self->{long_pack} C C",
	433	substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ),
	434	);
	435
	436	return 1 if $trans_id == $args->{trans_id};
	437	}
	438
	439	return;
	440	}
	441
	442	sub _remove_key_offset {
	443	my $self = shift;
	444	my ($args) = @_;
	445
	446	my $is_changed;
	447	for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
	448	my ($loc, $trans_id, $is_deleted) = unpack(
	449	"$self->{long_pack} C C",
	450	substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ),
	451	);
	452
	453	if ( $trans_id == $args->{trans_id} ) {
	454	substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ) = '';
	455	$args->{tag}{content} .= chr(0) x $self->{key_size};
	456	$is_changed = 1;
	457	redo;
	458	}
	459	}
	460
	461	if ( $is_changed ) {
	462	$self->_storage->print_at(
	463	$args->{tag}{offset}, $args->{tag}{content},
	464	);
	465	}
	466
	467	return 1;
	468	}
	469
40956c06	470	sub _add_key_offset {
fb451ba6	471	my $self = shift;
	472	my ($args) = @_;
	473
40956c06	474	my $is_changed;
	475	for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
	476	my ($loc, $trans_id, $is_deleted) = unpack(
	477	"$self->{long_pack} C C",
	478	substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ),
	479	);
fb451ba6	480
40956c06	481	if ( $trans_id == $args->{trans_id} \|\| (!$loc && !$is_deleted) ) {
	482	substr( $args->{tag}{content}, $i * $self->{key_size}, $self->{key_size} ) = pack(
	483	"$self->{long_pack} C C",
	484	$args->{loc}, $args->{trans_id}, 0,
	485	);
	486	$is_changed = 1;
	487	last;
	488	}
	489	}
	490
	491	if ( $is_changed ) {
	492	$self->_storage->print_at(
	493	$args->{tag}{offset}, $args->{tag}{content},
	494	);
	495	}
	496	else {
	497	die "Why didn't _add_key_offset() change something?!\n";
	498	}
	499
	500	return 1;
fb451ba6	501	}
	502
	503	sub setup_fh {
	504	my $self = shift;
	505	my ($obj) = @_;
	506
	507	# Need to remove use of $fh here
	508	my $fh = $self->_storage->{fh};
	509	flock $fh, LOCK_EX;
	510
	511	#XXX The duplication of calculate_sizes needs to go away
	512	unless ( $obj->{base_offset} ) {
	513	my $bytes_read = $self->read_file_header;
	514
	515	$self->calculate_sizes;
	516
	517	##
	518	# File is empty -- write header and master index
	519	##
	520	if (!$bytes_read) {
	521	$self->_storage->audit( "# Database created on" );
	522
	523	$self->write_file_header;
	524
	525	$obj->{base_offset} = $self->_storage->request_space(
	526	$self->tag_size( $self->{keyloc_size} ),
	527	);
	528
	529	my $value_spot = $self->_storage->request_space(
	530	$self->tag_size( $self->{index_size} ),
	531	);
	532
	533	$self->write_tag(
	534	$obj->{base_offset}, SIG_KEYS,
129ea236	535	pack( "$self->{long_pack} C C", $value_spot, HEAD, 0 ),
129ea236	536	chr(0) x ($self->{index_size} - $self->{key_size}),
fb451ba6	537	);
	538
	539	$self->write_tag(
	540	$value_spot, $obj->_type,
	541	chr(0)x$self->{index_size},
	542	);
	543
	544	# Flush the filehandle
	545	my $old_fh = select $fh;
	546	my $old_af = $\|; $\| = 1; $\| = $old_af;
	547	select $old_fh;
	548	}
	549	else {
	550	$obj->{base_offset} = $bytes_read;
	551
129ea236	552	my ($_val_offset, $_is_del) = $self->_find_value_offset({
	553	offset => $obj->{base_offset},
	554	trans_id => HEAD,
	555	allow_head => 1,
	556	});
	557	die "Attempt to use a deleted value" if $_is_del;
	558	die "Internal error!" if !$_val_offset;
	559
fb451ba6	560	##
	561	# Get our type from master index header
	562	##
129ea236	563	my $tag = $self->load_tag($_val_offset);
fb451ba6	564	unless ( $tag ) {
	565	flock $fh, LOCK_UN;
	566	$self->_throw_error("Corrupted file, no master index record");
	567	}
	568
	569	unless ($obj->_type eq $tag->{signature}) {
	570	flock $fh, LOCK_UN;
	571	$self->_throw_error("File type mismatch");
	572	}
	573	}
	574	}
	575	else {
	576	$self->calculate_sizes;
	577	}
	578
	579	#XXX We have to make sure we don't mess up when autoflush isn't turned on
	580	$self->_storage->set_inode;
	581
	582	flock $fh, LOCK_UN;
	583
	584	return 1;
	585	}
	586
	587	1;
	588	__END__