lib/DBM/Deep/Engine.pm

   1 package DBM::Deep::Engine;
   2
   3 use 5.6.0;
   4
   5 use strict;
   6
   7 our $VERSION = q(0.99_03);
   8
   9 use Scalar::Util ();
  10
  11 # File-wide notes:
  12 # * Every method in here assumes that the storage has been appropriately
  13 #   safeguarded. This can be anything from flock() to some sort of manual
  14 #   mutex. But, it's the caller's responsability to make sure that this has
  15 #   been done.
  16
  17 # Setup file and tag signatures.  These should never change.
  18 sub SIG_FILE     () { 'DPDB' }
  19 sub SIG_HEADER   () { 'h'    }
  20 sub SIG_INTERNAL () { 'i'    }
  21 sub SIG_HASH     () { 'H'    }
  22 sub SIG_ARRAY    () { 'A'    }
  23 sub SIG_NULL     () { 'N'    }
  24 sub SIG_DATA     () { 'D'    }
  25 sub SIG_INDEX    () { 'I'    }
  26 sub SIG_BLIST    () { 'B'    }
  27 sub SIG_FREE     () { 'F'    }
  28 sub SIG_KEYS     () { 'K'    }
  29 sub SIG_SIZE     () {  1     }
  30 sub STALE_SIZE   () {  1     }
  31
  32 ################################################################################
  33
  34 # Please refer to the pack() documentation for further information
  35 my %StP = (
  36     1 => 'C', # Unsigned char value (no order specified, presumably ASCII)
  37     2 => 'n', # Unsigned short in "network" (big-endian) order
  38     4 => 'N', # Unsigned long in "network" (big-endian) order
  39     8 => 'Q', # Usigned quad (no order specified, presumably machine-dependent)
  40 );
  41
  42 sub new {
  43     my $class = shift;
  44     my ($args) = @_;
  45
  46     my $self = bless {
  47         byte_size   => 4,
  48
  49         digest      => undef,
  50         hash_size   => 16,  # In bytes
  51         hash_chars  => 256, # Number of chars the algorithm uses per byte
  52         max_buckets => 16,
  53         num_txns    => 16, # HEAD plus 15 running txns
  54         trans_id    => 0,  # Default to the HEAD
  55
  56         entries => {}, # This is the list of entries for transactions
  57         storage => undef,
  58     }, $class;
  59
  60     if ( defined $args->{pack_size} ) {
  61         if ( lc $args->{pack_size} eq 'small' ) {
  62             $args->{byte_size} = 2;
  63         }
  64         elsif ( lc $args->{pack_size} eq 'medium' ) {
  65             $args->{byte_size} = 4;
  66         }
  67         elsif ( lc $args->{pack_size} eq 'large' ) {
  68             $args->{byte_size} = 8;
  69         }
  70         else {
  71             die "Unknown pack_size value: '$args->{pack_size}'\n";
  72         }
  73     }
  74
  75     # Grab the parameters we want to use
  76     foreach my $param ( keys %$self ) {
  77         next unless exists $args->{$param};
  78         $self->{$param} = $args->{$param};
  79     }
  80
  81     $self->{byte_pack} = $StP{ $self->byte_size };
  82
  83     ##
  84     # Number of buckets per blist before another level of indexing is
  85     # done. Increase this value for slightly greater speed, but larger database
  86     # files. DO NOT decrease this value below 16, due to risk of recursive
  87     # reindex overrun.
  88     ##
  89     if ( $self->{max_buckets} < 16 ) {
  90         warn "Floor of max_buckets is 16. Setting it to 16 from '$self->{max_buckets}'\n";
  91         $self->{max_buckets} = 16;
  92     }
  93
  94     if ( !$self->{digest} ) {
  95         require Digest::MD5;
  96         $self->{digest} = \&Digest::MD5::md5;
  97     }
  98
  99     return $self;
 100 }
 101
 102 ################################################################################
 103
 104 sub read_value {
 105     my $self = shift;
 106     my ($obj, $key) = @_;
 107
 108     # This will be a Reference sector
 109     my $sector = $self->_load_sector( $obj->_base_offset )
 110         or return;
 111
 112     if ( $sector->staleness != $obj->_staleness ) {
 113         return;
 114     }
 115
 116     my $key_md5 = $self->_apply_digest( $key );
 117
 118     my $value_sector = $sector->get_data_for({
 119         key_md5    => $key_md5,
 120         allow_head => 1,
 121     });
 122
 123     unless ( $value_sector ) {
 124         $value_sector = DBM::Deep::Engine::Sector::Null->new({
 125             engine => $self,
 126             data   => undef,
 127         });
 128
 129         $sector->write_data({
 130             key_md5 => $key_md5,
 131             key     => $key,
 132             value   => $value_sector,
 133         });
 134     }
 135
 136     return $value_sector->data;
 137 }
 138
 139 sub get_classname {
 140     my $self = shift;
 141     my ($obj) = @_;
 142
 143     # This will be a Reference sector
 144     my $sector = $self->_load_sector( $obj->_base_offset )
 145         or die "How did get_classname fail (no sector for '$obj')?!\n";
 146
 147     if ( $sector->staleness != $obj->_staleness ) {
 148         return;
 149     }
 150
 151     return $sector->get_classname;
 152 }
 153
 154 sub key_exists {
 155     my $self = shift;
 156     my ($obj, $key) = @_;
 157
 158     # This will be a Reference sector
 159     my $sector = $self->_load_sector( $obj->_base_offset )
 160         or return '';
 161
 162     if ( $sector->staleness != $obj->_staleness ) {
 163         return '';
 164     }
 165
 166     my $data = $sector->get_data_for({
 167         key_md5    => $self->_apply_digest( $key ),
 168         allow_head => 1,
 169     });
 170
 171     # exists() returns 1 or '' for true/false.
 172     return $data ? 1 : '';
 173 }
 174
 175 sub delete_key {
 176     my $self = shift;
 177     my ($obj, $key) = @_;
 178
 179     my $sector = $self->_load_sector( $obj->_base_offset )
 180         or return;
 181
 182     if ( $sector->staleness != $obj->_staleness ) {
 183         return;
 184     }
 185
 186     return $sector->delete_key({
 187         key_md5    => $self->_apply_digest( $key ),
 188         allow_head => 0,
 189     });
 190 }
 191
 192 sub write_value {
 193     my $self = shift;
 194     my ($obj, $key, $value) = @_;
 195
 196     my $r = Scalar::Util::reftype( $value ) || '';
 197     {
 198         last if $r eq '';
 199         last if $r eq 'HASH';
 200         last if $r eq 'ARRAY';
 201
 202         DBM::Deep->_throw_error(
 203             "Storage of references of type '$r' is not supported."
 204         );
 205     }
 206
 207     my ($class, $type);
 208     if ( !defined $value ) {
 209         $class = 'DBM::Deep::Engine::Sector::Null';
 210     }
 211     elsif ( $r eq 'ARRAY' || $r eq 'HASH' ) {
 212         if ( $r eq 'ARRAY' && tied(@$value) ) {
 213             DBM::Deep->_throw_error( "Cannot store something that is tied." );
 214         }
 215         if ( $r eq 'HASH' && tied(%$value) ) {
 216             DBM::Deep->_throw_error( "Cannot store something that is tied." );
 217         }
 218         $class = 'DBM::Deep::Engine::Sector::Reference';
 219         $type = substr( $r, 0, 1 );
 220     }
 221     else {
 222         $class = 'DBM::Deep::Engine::Sector::Scalar';
 223     }
 224
 225     # This will be a Reference sector
 226     my $sector = $self->_load_sector( $obj->_base_offset )
 227         or die "Cannot write to a deleted spot in DBM::Deep.\n";
 228
 229     if ( $sector->staleness != $obj->_staleness ) {
 230         die "Cannot write to a deleted spot in DBM::Deep.\n";
 231     }
 232
 233     # Create this after loading the reference sector in case something bad happens.
 234     # This way, we won't allocate value sector(s) needlessly.
 235     my $value_sector = $class->new({
 236         engine => $self,
 237         data   => $value,
 238         type   => $type,
 239     });
 240
 241     $sector->write_data({
 242         key     => $key,
 243         key_md5 => $self->_apply_digest( $key ),
 244         value   => $value_sector,
 245     });
 246
 247     # This code is to make sure we write all the values in the $value to the disk
 248     # and to make sure all changes to $value after the assignment are reflected
 249     # on disk. This may be counter-intuitive at first, but it is correct dwimmery.
 250     #   NOTE - simply tying $value won't perform a STORE on each value. Hence, the
 251     # copy to a temp value.
 252     if ( $r eq 'ARRAY' ) {
 253         my @temp = @$value;
 254         tie @$value, 'DBM::Deep', {
 255             base_offset => $value_sector->offset,
 256             staleness   => $value_sector->staleness,
 257             storage     => $self->storage,
 258             engine      => $self,
 259         };
 260         @$value = @temp;
 261         bless $value, 'DBM::Deep::Array' unless Scalar::Util::blessed( $value );
 262     }
 263     elsif ( $r eq 'HASH' ) {
 264         my %temp = %$value;
 265         tie %$value, 'DBM::Deep', {
 266             base_offset => $value_sector->offset,
 267             staleness   => $value_sector->staleness,
 268             storage     => $self->storage,
 269             engine      => $self,
 270         };
 271
 272         %$value = %temp;
 273         bless $value, 'DBM::Deep::Hash' unless Scalar::Util::blessed( $value );
 274     }
 275
 276     return 1;
 277 }
 278
 279 # XXX Add staleness here
 280 sub get_next_key {
 281     my $self = shift;
 282     my ($obj, $prev_key) = @_;
 283
 284     # XXX Need to add logic about resetting the iterator if any key in the reference has changed
 285     unless ( $prev_key ) {
 286         $obj->{iterator} = DBM::Deep::Iterator->new({
 287             base_offset => $obj->_base_offset,
 288             engine      => $self,
 289         });
 290     }
 291
 292     return $obj->{iterator}->get_next_key( $obj );
 293 }
 294
 295 ################################################################################
 296
 297 sub setup_fh {
 298     my $self = shift;
 299     my ($obj) = @_;
 300
 301     # We're opening the file.
 302     unless ( $obj->_base_offset ) {
 303         my $bytes_read = $self->_read_file_header;
 304
 305         # Creating a new file
 306         unless ( $bytes_read ) {
 307             $self->_write_file_header;
 308
 309             # 1) Create Array/Hash entry
 310             my $initial_reference = DBM::Deep::Engine::Sector::Reference->new({
 311                 engine => $self,
 312                 type   => $obj->_type,
 313             });
 314             $obj->{base_offset} = $initial_reference->offset;
 315             $obj->{staleness} = $initial_reference->staleness;
 316
 317             $self->storage->flush;
 318         }
 319         # Reading from an existing file
 320         else {
 321             $obj->{base_offset} = $bytes_read;
 322             my $initial_reference = DBM::Deep::Engine::Sector::Reference->new({
 323                 engine => $self,
 324                 offset => $obj->_base_offset,
 325             });
 326             unless ( $initial_reference ) {
 327                 DBM::Deep->_throw_error("Corrupted file, no master index record");
 328             }
 329
 330             unless ($obj->_type eq $initial_reference->type) {
 331                 DBM::Deep->_throw_error("File type mismatch");
 332             }
 333
 334             $obj->{staleness} = $initial_reference->staleness;
 335         }
 336     }
 337
 338     return 1;
 339 }
 340
 341 sub begin_work {
 342     my $self = shift;
 343     my ($obj) = @_;
 344
 345     if ( $self->trans_id ) {
 346         DBM::Deep->_throw_error( "Cannot begin_work within a transaction" );
 347     }
 348
 349     my @slots = $self->read_txn_slots;
 350     for my $i ( 1 .. @slots ) {
 351         next if $slots[$i];
 352         $slots[$i] = 1;
 353         $self->set_trans_id( $i );
 354         last;
 355     }
 356     $self->write_txn_slots( @slots );
 357
 358     if ( !$self->trans_id ) {
 359         DBM::Deep->_throw_error( "Cannot begin_work - no available transactions" );
 360     }
 361
 362     return;
 363 }
 364
 365 sub rollback {
 366     my $self = shift;
 367     my ($obj) = @_;
 368
 369     if ( !$self->trans_id ) {
 370         DBM::Deep->_throw_error( "Cannot rollback without a transaction" );
 371     }
 372
 373     # Each entry is the file location for a bucket that has a modification for
 374     # this transaction. The entries need to be expunged.
 375     foreach my $entry (@{ $self->get_entries } ) {
 376         # Remove the entry here
 377         my $read_loc = $entry
 378           + $self->hash_size
 379           + $self->byte_size
 380           + $self->trans_id * ( $self->byte_size + 4 );
 381
 382         my $data_loc = $self->storage->read_at( $read_loc, $self->byte_size );
 383         $data_loc = unpack( $StP{$self->byte_size}, $data_loc );
 384         $self->storage->print_at( $read_loc, pack( $StP{$self->byte_size}, 0 ) );
 385
 386         if ( $data_loc > 1 ) {
 387             $self->_load_sector( $data_loc )->free;
 388         }
 389     }
 390
 391     $self->clear_entries;
 392
 393     my @slots = $self->read_txn_slots;
 394     $slots[$self->trans_id] = 0;
 395     $self->write_txn_slots( @slots );
 396     $self->inc_txn_staleness_counter( $self->trans_id );
 397     $self->set_trans_id( 0 );
 398
 399     return 1;
 400 }
 401
 402 sub commit {
 403     my $self = shift;
 404     my ($obj) = @_;
 405
 406     if ( !$self->trans_id ) {
 407         DBM::Deep->_throw_error( "Cannot commit without a transaction" );
 408     }
 409
 410     foreach my $entry (@{ $self->get_entries } ) {
 411         # Overwrite the entry in head with the entry in trans_id
 412         my $base = $entry
 413           + $self->hash_size
 414           + $self->byte_size;
 415
 416         my $head_loc = $self->storage->read_at( $base, $self->byte_size );
 417         $head_loc = unpack( $StP{$self->byte_size}, $head_loc );
 418         my $trans_loc = $self->storage->read_at(
 419             $base + $self->trans_id * ( $self->byte_size + 4 ), $self->byte_size,
 420         );
 421
 422         $self->storage->print_at( $base, $trans_loc );
 423         $self->storage->print_at(
 424             $base + $self->trans_id * ( $self->byte_size + 4 ),
 425             pack( $StP{$self->byte_size} . ' N', (0) x 2 ),
 426         );
 427
 428         if ( $head_loc > 1 ) {
 429             $self->_load_sector( $head_loc )->free;
 430         }
 431     }
 432
 433     $self->clear_entries;
 434
 435     my @slots = $self->read_txn_slots;
 436     $slots[$self->trans_id] = 0;
 437     $self->write_txn_slots( @slots );
 438     $self->inc_txn_staleness_counter( $self->trans_id );
 439     $self->set_trans_id( 0 );
 440
 441     return 1;
 442 }
 443
 444 sub read_txn_slots {
 445     my $self = shift;
 446     return split '', unpack( 'b32',
 447         $self->storage->read_at(
 448             $self->trans_loc, 4,
 449         )
 450     );
 451 }
 452
 453 sub write_txn_slots {
 454     my $self = shift;
 455     $self->storage->print_at( $self->trans_loc,
 456         pack( 'b32', join('', @_) ),
 457     );
 458 }
 459
 460 sub get_running_txn_ids {
 461     my $self = shift;
 462     my @transactions = $self->read_txn_slots;
 463     my @trans_ids = grep { $transactions[$_] } 0 .. $#transactions;
 464 }
 465
 466 sub get_txn_staleness_counter {
 467     my $self = shift;
 468     my ($trans_id) = @_;
 469
 470     # Hardcode staleness of 0 for the HEAD
 471     return 0 unless $trans_id;
 472
 473     my $x = unpack( 'N',
 474         $self->storage->read_at(
 475             $self->trans_loc + 4 * $trans_id,
 476             4,
 477         )
 478     );
 479     return $x;
 480 }
 481
 482 sub inc_txn_staleness_counter {
 483     my $self = shift;
 484     my ($trans_id) = @_;
 485
 486     # Hardcode staleness of 0 for the HEAD
 487     return unless $trans_id;
 488
 489     $self->storage->print_at(
 490         $self->trans_loc + 4 * $trans_id,
 491         pack( 'N', $self->get_txn_staleness_counter( $trans_id ) + 1 ),
 492     );
 493 }
 494
 495 sub get_entries {
 496     my $self = shift;
 497     return [ keys %{ $self->{entries}{$self->trans_id} ||= {} } ];
 498 }
 499
 500 sub add_entry {
 501     my $self = shift;
 502     my ($trans_id, $loc) = @_;
 503
 504     $self->{entries}{$trans_id} ||= {};
 505     $self->{entries}{$trans_id}{$loc} = undef;
 506 }
 507
 508 # If the buckets are being relocated because of a reindexing, the entries
 509 # mechanism needs to be made aware of it.
 510 sub reindex_entry {
 511     my $self = shift;
 512     my ($old_loc, $new_loc) = @_;
 513
 514     TRANS:
 515     while ( my ($trans_id, $locs) = each %{ $self->{entries} } ) {
 516         foreach my $orig_loc ( keys %{ $locs } ) {
 517             if ( $orig_loc == $old_loc ) {
 518                 delete $locs->{orig_loc};
 519                 $locs->{$new_loc} = undef;
 520                 next TRANS;
 521             }
 522         }
 523     }
 524 }
 525
 526 sub clear_entries {
 527     my $self = shift;
 528     delete $self->{entries}{$self->trans_id};
 529 }
 530
 531 ################################################################################
 532
 533 {
 534     my $header_fixed = length( SIG_FILE ) + 1 + 4 + 4;
 535
 536     sub _write_file_header {
 537         my $self = shift;
 538
 539         my $header_var = 1 + 1 + 4 + 4 * $self->num_txns + 3 * $self->byte_size;
 540
 541         my $loc = $self->storage->request_space( $header_fixed + $header_var );
 542
 543         $self->storage->print_at( $loc,
 544             SIG_FILE,
 545             SIG_HEADER,
 546             pack('N', 1),           # header version - at this point, we're at 9 bytes
 547             pack('N', $header_var), # header size
 548             # --- Above is $header_fixed. Below is $header_var
 549             pack('C', $self->byte_size),
 550             pack('C', $self->max_buckets),
 551             pack('N', 0 ),                   # Transaction activeness bitfield
 552             pack('N' . $self->num_txns, 0 x $self->num_txns ), # Transaction staleness counters
 553             pack($StP{$self->byte_size}, 0), # Start of free chain (blist size)
 554             pack($StP{$self->byte_size}, 0), # Start of free chain (data size)
 555             pack($StP{$self->byte_size}, 0), # Start of free chain (index size)
 556         );
 557
 558         $self->set_trans_loc( $header_fixed + 2 );
 559         $self->set_chains_loc( $header_fixed + 2 + 4 + 4 * $self->num_txns );
 560
 561         return;
 562     }
 563
 564     sub _read_file_header {
 565         my $self = shift;
 566
 567         my $buffer = $self->storage->read_at( 0, $header_fixed );
 568         return unless length($buffer);
 569
 570         my ($file_signature, $sig_header, $header_version, $size) = unpack(
 571             'A4 A N N', $buffer
 572         );
 573
 574         unless ( $file_signature eq SIG_FILE ) {
 575             $self->storage->close;
 576             DBM::Deep->_throw_error( "Signature not found -- file is not a Deep DB" );
 577         }
 578
 579         unless ( $sig_header eq SIG_HEADER ) {
 580             $self->storage->close;
 581             DBM::Deep->_throw_error( "Old file version found." );
 582         }
 583
 584         my $buffer2 = $self->storage->read_at( undef, $size );
 585         my @values = unpack( 'C C', $buffer2 );
 586
 587         $self->set_trans_loc( $header_fixed + 2 );
 588         $self->set_chains_loc( $header_fixed + 2 + 4 + 4 * $self->num_txns );
 589
 590         if ( @values < 2 || grep { !defined } @values ) {
 591             $self->storage->close;
 592             DBM::Deep->_throw_error("Corrupted file - bad header");
 593         }
 594
 595         #XXX Add warnings if values weren't set right
 596         @{$self}{qw(byte_size max_buckets)} = @values;
 597
 598         my $header_var = 1 + 1 + 4 + 4 * $self->num_txns + 3 * $self->byte_size;
 599         unless ( $size eq $header_var ) {
 600             $self->storage->close;
 601             DBM::Deep->_throw_error( "Unexpected size found ($size <-> $header_var)." );
 602         }
 603
 604         return length($buffer) + length($buffer2);
 605     }
 606 }
 607
 608 sub _load_sector {
 609     my $self = shift;
 610     my ($offset) = @_;
 611
 612     # Add a catch for offset of 0 or 1
 613     return if $offset <= 1;
 614
 615     my $type = $self->storage->read_at( $offset, 1 );
 616     return if $type eq chr(0);
 617
 618     if ( $type eq $self->SIG_ARRAY || $type eq $self->SIG_HASH ) {
 619         return DBM::Deep::Engine::Sector::Reference->new({
 620             engine => $self,
 621             type   => $type,
 622             offset => $offset,
 623         });
 624     }
 625     # XXX Don't we need key_md5 here?
 626     elsif ( $type eq $self->SIG_BLIST ) {
 627         return DBM::Deep::Engine::Sector::BucketList->new({
 628             engine => $self,
 629             type   => $type,
 630             offset => $offset,
 631         });
 632     }
 633     elsif ( $type eq $self->SIG_INDEX ) {
 634         return DBM::Deep::Engine::Sector::Index->new({
 635             engine => $self,
 636             type   => $type,
 637             offset => $offset,
 638         });
 639     }
 640     elsif ( $type eq $self->SIG_NULL ) {
 641         return DBM::Deep::Engine::Sector::Null->new({
 642             engine => $self,
 643             type   => $type,
 644             offset => $offset,
 645         });
 646     }
 647     elsif ( $type eq $self->SIG_DATA ) {
 648         return DBM::Deep::Engine::Sector::Scalar->new({
 649             engine => $self,
 650             type   => $type,
 651             offset => $offset,
 652         });
 653     }
 654     # This was deleted from under us, so just return and let the caller figure it out.
 655     elsif ( $type eq $self->SIG_FREE ) {
 656         return;
 657     }
 658
 659     die "'$offset': Don't know what to do with type '$type'\n";
 660 }
 661
 662 sub _apply_digest {
 663     my $self = shift;
 664     return $self->{digest}->(@_);
 665 }
 666
 667 sub _add_free_blist_sector { shift->_add_free_sector( 0, @_ ) }
 668 sub _add_free_data_sector { shift->_add_free_sector( 1, @_ ) }
 669 sub _add_free_index_sector { shift->_add_free_sector( 2, @_ ) }
 670
 671 sub _add_free_sector {
 672     my $self = shift;
 673     my ($multiple, $offset, $size) = @_;
 674
 675     my $chains_offset = $multiple * $self->byte_size;
 676
 677     my $storage = $self->storage;
 678
 679     # Increment staleness.
 680     # XXX Can this increment+modulo be done by "&= 0x1" ?
 681     my $staleness = unpack( $StP{STALE_SIZE()}, $storage->read_at( $offset + SIG_SIZE, STALE_SIZE ) );
 682     $staleness = ($staleness + 1 ) % ( 2 ** ( 8 * STALE_SIZE ) );
 683     $storage->print_at( $offset + SIG_SIZE, pack( $StP{STALE_SIZE()}, $staleness ) );
 684
 685     my $old_head = $storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 686
 687     $storage->print_at( $self->chains_loc + $chains_offset,
 688         pack( $StP{$self->byte_size}, $offset ),
 689     );
 690
 691     # Record the old head in the new sector after the signature and staleness counter
 692     $storage->print_at( $offset + SIG_SIZE + STALE_SIZE, $old_head );
 693 }
 694
 695 sub _request_blist_sector { shift->_request_sector( 0, @_ ) }
 696 sub _request_data_sector { shift->_request_sector( 1, @_ ) }
 697 sub _request_index_sector { shift->_request_sector( 2, @_ ) }
 698
 699 sub _request_sector {
 700     my $self = shift;
 701     my ($multiple, $size) = @_;
 702
 703     my $chains_offset = $multiple * $self->byte_size;
 704
 705     my $old_head = $self->storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 706     my $loc = unpack( $StP{$self->byte_size}, $old_head );
 707
 708     # We don't have any free sectors of the right size, so allocate a new one.
 709     unless ( $loc ) {
 710         my $offset = $self->storage->request_space( $size );
 711
 712         # Zero out the new sector. This also guarantees correct increases
 713         # in the filesize.
 714         $self->storage->print_at( $offset, chr(0) x $size );
 715
 716         return $offset;
 717     }
 718
 719     # Read the new head after the signature and the staleness counter
 720     my $new_head = $self->storage->read_at( $loc + SIG_SIZE + STALE_SIZE, $self->byte_size );
 721     $self->storage->print_at( $self->chains_loc + $chains_offset, $new_head );
 722
 723     return $loc;
 724 }
 725
 726 ################################################################################
 727
 728 sub storage     { $_[0]{storage} }
 729 sub byte_size   { $_[0]{byte_size} }
 730 sub hash_size   { $_[0]{hash_size} }
 731 sub hash_chars  { $_[0]{hash_chars} }
 732 sub num_txns    { $_[0]{num_txns} }
 733 sub max_buckets { $_[0]{max_buckets} }
 734 sub blank_md5   { chr(0) x $_[0]->hash_size }
 735
 736 sub trans_id     { $_[0]{trans_id} }
 737 sub set_trans_id { $_[0]{trans_id} = $_[1] }
 738
 739 sub trans_loc     { $_[0]{trans_loc} }
 740 sub set_trans_loc { $_[0]{trans_loc} = $_[1] }
 741
 742 sub chains_loc     { $_[0]{chains_loc} }
 743 sub set_chains_loc { $_[0]{chains_loc} = $_[1] }
 744
 745 ################################################################################
 746
 747 package DBM::Deep::Iterator;
 748
 749 sub new {
 750     my $class = shift;
 751     my ($args) = @_;
 752
 753     my $self = bless {
 754         breadcrumbs => [],
 755         engine      => $args->{engine},
 756         base_offset => $args->{base_offset},
 757     }, $class;
 758
 759     Scalar::Util::weaken( $self->{engine} );
 760
 761     return $self;
 762 }
 763
 764 sub reset { $_[0]{breadcrumbs} = [] }
 765
 766 sub get_sector_iterator {
 767     my $self = shift;
 768     my ($loc) = @_;
 769
 770     my $sector = $self->{engine}->_load_sector( $loc )
 771         or return;
 772
 773     if ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
 774         return DBM::Deep::Iterator::Index->new({
 775             iterator => $self,
 776             sector   => $sector,
 777         });
 778     }
 779     elsif ( $sector->isa( 'DBM::Deep::Engine::Sector::BucketList' ) ) {
 780         return DBM::Deep::Iterator::BucketList->new({
 781             iterator => $self,
 782             sector   => $sector,
 783         });
 784     }
 785     else {
 786         die "Why did $loc make a $sector?";
 787     }
 788 }
 789
 790 sub get_next_key {
 791     my $self = shift;
 792     my ($obj) = @_;
 793
 794     my $crumbs = $self->{breadcrumbs};
 795     my $e = $self->{engine};
 796
 797     unless ( @$crumbs ) {
 798         # This will be a Reference sector
 799         my $sector = $e->_load_sector( $self->{base_offset} )
 800             # If no sector is found, thist must have been deleted from under us.
 801             or return;
 802
 803         if ( $sector->staleness != $obj->_staleness ) {
 804             return;
 805         }
 806
 807         my $loc = $sector->get_blist_loc
 808             or return;
 809
 810         push @$crumbs, $self->get_sector_iterator( $loc );
 811     }
 812
 813     FIND_NEXT_KEY: {
 814         # We're at the end.
 815         unless ( @$crumbs ) {
 816             $self->reset;
 817             return;
 818         }
 819
 820         my $iterator = $crumbs->[-1];
 821
 822         # This level is done.
 823         if ( $iterator->at_end ) {
 824             pop @$crumbs;
 825             redo FIND_NEXT_KEY;
 826         }
 827
 828         if ( $iterator->isa( 'DBM::Deep::Iterator::Index' ) ) {
 829             # If we don't have any more, it will be caught at the
 830             # prior check.
 831             if ( my $next = $iterator->get_next_iterator ) {
 832                 push @$crumbs, $next;
 833             }
 834             redo FIND_NEXT_KEY;
 835         }
 836
 837         unless ( $iterator->isa( 'DBM::Deep::Iterator::BucketList' ) ) {
 838             DBM::Deep->_throw_error(
 839                 "Should have a bucketlist iterator here - instead have $iterator"
 840             );
 841         }
 842
 843         # At this point, we have a BucketList iterator
 844         my $key = $iterator->get_next_key;
 845         if ( defined $key ) {
 846             return $key;
 847         }
 848
 849         # We hit the end of the bucketlist iterator, so redo
 850         redo FIND_NEXT_KEY;
 851     }
 852
 853     DBM::Deep->_throw_error( "get_next_key(): How did we get here?" );
 854 }
 855
 856 package DBM::Deep::Iterator::Index;
 857
 858 sub new {
 859     my $self = bless $_[1] => $_[0];
 860     $self->{curr_index} = 0;
 861     return $self;
 862 }
 863
 864 sub at_end {
 865     my $self = shift;
 866     return $self->{curr_index} >= $self->{iterator}{engine}->hash_chars;
 867 }
 868
 869 sub get_next_iterator {
 870     my $self = shift;
 871
 872     my $loc;
 873     while ( !$loc ) {
 874         return if $self->at_end;
 875         $loc = $self->{sector}->get_entry( $self->{curr_index}++ );
 876     }
 877
 878     return $self->{iterator}->get_sector_iterator( $loc );
 879 }
 880
 881 package DBM::Deep::Iterator::BucketList;
 882
 883 sub new {
 884     my $self = bless $_[1] => $_[0];
 885     $self->{curr_index} = 0;
 886     return $self;
 887 }
 888
 889 sub at_end {
 890     my $self = shift;
 891     return $self->{curr_index} >= $self->{iterator}{engine}->max_buckets;
 892 }
 893
 894 sub get_next_key {
 895     my $self = shift;
 896
 897     return if $self->at_end;
 898
 899     my $data_loc = $self->{sector}->get_data_location_for({
 900         allow_head => 1,
 901         idx => $self->{curr_index}++,
 902     }) or return;
 903
 904     my $key_sector = $self->{sector}->get_key_for( $self->{curr_index} - 1 );
 905
 906     #XXX Is this check necessary now?
 907     return unless $key_sector;
 908
 909     return $key_sector->data;
 910 }
 911
 912 package DBM::Deep::Engine::Sector;
 913
 914 sub new {
 915     my $self = bless $_[1], $_[0];
 916     Scalar::Util::weaken( $self->{engine} );
 917     $self->_init;
 918     return $self;
 919 }
 920 sub _init {}
 921 sub clone { die "Must be implemented in the child class" }
 922
 923 sub engine { $_[0]{engine} }
 924 sub offset { $_[0]{offset} }
 925 sub type   { $_[0]{type} }
 926
 927 sub base_size {
 928    my $self = shift;
 929    return $self->engine->SIG_SIZE + $self->engine->STALE_SIZE;
 930 }
 931
 932 sub free {
 933     my $self = shift;
 934
 935     my $e = $self->engine;
 936
 937     $e->storage->print_at( $self->offset, $e->SIG_FREE );
 938     # Skip staleness counter
 939     $e->storage->print_at( $self->offset + $self->base_size,
 940         chr(0) x ($self->size - $self->base_size),
 941     );
 942
 943     my $free_meth = $self->free_meth;
 944     $e->$free_meth( $self->offset, $self->size );
 945
 946     return;
 947 }
 948
 949 package DBM::Deep::Engine::Sector::Data;
 950
 951 our @ISA = qw( DBM::Deep::Engine::Sector );
 952
 953 # This is in bytes
 954 sub size { return 256 }
 955 sub free_meth { return '_add_free_data_sector' }
 956
 957 sub clone {
 958     my $self = shift;
 959     return ref($self)->new({
 960         engine => $self->engine,
 961         data   => $self->data,
 962         type   => $self->type,
 963     });
 964 }
 965
 966 package DBM::Deep::Engine::Sector::Scalar;
 967
 968 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
 969
 970 sub free {
 971     my $self = shift;
 972
 973     my $chain_loc = $self->chain_loc;
 974
 975     $self->SUPER::free();
 976
 977     if ( $chain_loc ) {
 978         $self->engine->_load_sector( $chain_loc )->free;
 979     }
 980
 981     return;
 982 }
 983
 984 sub type { $_[0]{engine}->SIG_DATA }
 985 sub _init {
 986     my $self = shift;
 987
 988     my $engine = $self->engine;
 989
 990     unless ( $self->offset ) {
 991         my $data_section = $self->size - $self->base_size - 1 * $engine->byte_size - 1;
 992
 993         $self->{offset} = $engine->_request_data_sector( $self->size );
 994
 995         my $data = delete $self->{data};
 996         my $dlen = length $data;
 997         my $continue = 1;
 998         my $curr_offset = $self->offset;
 999         while ( $continue ) {
1000
1001             my $next_offset = 0;
1002
1003             my ($leftover, $this_len, $chunk);
1004             if ( $dlen > $data_section ) {
1005                 $leftover = 0;
1006                 $this_len = $data_section;
1007                 $chunk = substr( $data, 0, $this_len );
1008
1009                 $dlen -= $data_section;
1010                 $next_offset = $engine->_request_data_sector( $self->size );
1011                 $data = substr( $data, $this_len );
1012             }
1013             else {
1014                 $leftover = $data_section - $dlen;
1015                 $this_len = $dlen;
1016                 $chunk = $data;
1017
1018                 $continue = 0;
1019             }
1020
1021             $engine->storage->print_at( $curr_offset, $self->type ); # Sector type
1022             # Skip staleness
1023             $engine->storage->print_at( $curr_offset + $self->base_size,
1024                 pack( $StP{$engine->byte_size}, $next_offset ),  # Chain loc
1025                 pack( $StP{1}, $this_len ),                      # Data length
1026                 $chunk,                                          # Data to be stored in this sector
1027                 chr(0) x $leftover,                              # Zero-fill the rest
1028             );
1029
1030             $curr_offset = $next_offset;
1031         }
1032
1033         return;
1034     }
1035 }
1036
1037 sub data_length {
1038     my $self = shift;
1039
1040     my $buffer = $self->engine->storage->read_at(
1041         $self->offset + $self->base_size + $self->engine->byte_size, 1
1042     );
1043
1044     return unpack( $StP{1}, $buffer );
1045 }
1046
1047 sub chain_loc {
1048     my $self = shift;
1049     my $chain_loc = $self->engine->storage->read_at(
1050         $self->offset + $self->base_size, $self->engine->byte_size,
1051     );
1052     return unpack( $StP{$self->engine->byte_size}, $chain_loc );
1053 }
1054
1055 sub data {
1056     my $self = shift;
1057
1058     my $data;
1059     while ( 1 ) {
1060         my $chain_loc = $self->chain_loc;
1061
1062         $data .= $self->engine->storage->read_at(
1063             $self->offset + $self->base_size + $self->engine->byte_size + 1, $self->data_length,
1064         );
1065
1066         last unless $chain_loc;
1067
1068         $self = $self->engine->_load_sector( $chain_loc );
1069     }
1070
1071     return $data;
1072 }
1073
1074 package DBM::Deep::Engine::Sector::Null;
1075
1076 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1077
1078 sub type { $_[0]{engine}->SIG_NULL }
1079 sub data_length { 0 }
1080 sub data { return }
1081
1082 sub _init {
1083     my $self = shift;
1084
1085     my $engine = $self->engine;
1086
1087     unless ( $self->offset ) {
1088         my $leftover = $self->size - $self->base_size - 1 * $engine->byte_size - 1;
1089
1090         $self->{offset} = $engine->_request_data_sector( $self->size );
1091         $engine->storage->print_at( $self->offset, $self->type ); # Sector type
1092         # Skip staleness counter
1093         $engine->storage->print_at( $self->offset + $self->base_size,
1094             pack( $StP{$engine->byte_size}, 0 ),  # Chain loc
1095             pack( $StP{1}, $self->data_length ),  # Data length
1096             chr(0) x $leftover,                   # Zero-fill the rest
1097         );
1098
1099         return;
1100     }
1101 }
1102
1103 package DBM::Deep::Engine::Sector::Reference;
1104
1105 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1106
1107 sub _init {
1108     my $self = shift;
1109
1110     my $e = $self->engine;
1111
1112     unless ( $self->offset ) {
1113         my $classname = Scalar::Util::blessed( delete $self->{data} );
1114         my $leftover = $self->size - $self->base_size - 2 * $e->byte_size;
1115
1116         my $class_offset = 0;
1117         if ( defined $classname ) {
1118             my $class_sector = DBM::Deep::Engine::Sector::Scalar->new({
1119                 engine => $e,
1120                 data   => $classname,
1121             });
1122             $class_offset = $class_sector->offset;
1123         }
1124
1125         $self->{offset} = $e->_request_data_sector( $self->size );
1126         $e->storage->print_at( $self->offset, $self->type ); # Sector type
1127         # Skip staleness counter
1128         $e->storage->print_at( $self->offset + $self->base_size,
1129             pack( $StP{$e->byte_size}, 0 ),             # Index/BList loc
1130             pack( $StP{$e->byte_size}, $class_offset ), # Classname loc
1131             chr(0) x $leftover,                         # Zero-fill the rest
1132         );
1133     }
1134     else {
1135         $self->{type} = $e->storage->read_at( $self->offset, 1 );
1136     }
1137
1138     $self->{staleness} = unpack(
1139         $StP{$e->STALE_SIZE},
1140         $e->storage->read_at( $self->offset + $e->SIG_SIZE, $e->STALE_SIZE ),
1141     );
1142
1143     return;
1144 }
1145
1146 sub free {
1147     my $self = shift;
1148
1149     my $blist_loc = $self->get_blist_loc;
1150     $self->engine->_load_sector( $blist_loc )->free if $blist_loc;
1151
1152     my $class_loc = $self->get_class_offset;
1153     $self->engine->_load_sector( $class_loc )->free if $class_loc;
1154
1155     $self->SUPER::free();
1156 }
1157
1158 sub staleness { $_[0]{staleness} }
1159
1160 sub get_data_for {
1161     my $self = shift;
1162     my ($args) = @_;
1163
1164     # Assume that the head is not allowed unless otherwise specified.
1165     $args->{allow_head} = 0 unless exists $args->{allow_head};
1166
1167     # Assume we don't create a new blist location unless otherwise specified.
1168     $args->{create} = 0 unless exists $args->{create};
1169
1170     my $blist = $self->get_bucket_list({
1171         key_md5 => $args->{key_md5},
1172         key => $args->{key},
1173         create  => $args->{create},
1174     });
1175     return unless $blist && $blist->{found};
1176
1177     # At this point, $blist knows where the md5 is. What it -doesn't- know yet
1178     # is whether or not this transaction has this key. That's part of the next
1179     # function call.
1180     my $location = $blist->get_data_location_for({
1181         allow_head => $args->{allow_head},
1182     }) or return;
1183
1184     return $self->engine->_load_sector( $location );
1185 }
1186
1187 sub write_data {
1188     my $self = shift;
1189     my ($args) = @_;
1190
1191     my $blist = $self->get_bucket_list({
1192         key_md5 => $args->{key_md5},
1193         key => $args->{key},
1194         create  => 1,
1195     }) or die "How did write_data fail (no blist)?!\n";
1196
1197     # Handle any transactional bookkeeping.
1198     if ( $self->engine->trans_id ) {
1199         if ( ! $blist->has_md5 ) {
1200             $blist->mark_deleted({
1201                 trans_id => 0,
1202             });
1203         }
1204     }
1205     else {
1206         my @trans_ids = $self->engine->get_running_txn_ids;
1207         if ( $blist->has_md5 ) {
1208             if ( @trans_ids ) {
1209                 my $old_value = $blist->get_data_for;
1210                 foreach my $other_trans_id ( @trans_ids ) {
1211                     next if $blist->get_data_location_for({
1212                         trans_id   => $other_trans_id,
1213                         allow_head => 0,
1214                     });
1215                     $blist->write_md5({
1216                         trans_id => $other_trans_id,
1217                         key      => $args->{key},
1218                         key_md5  => $args->{key_md5},
1219                         value    => $old_value->clone,
1220                     });
1221                 }
1222             }
1223         }
1224         else {
1225             if ( @trans_ids ) {
1226                 foreach my $other_trans_id ( @trans_ids ) {
1227                     next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1228                     $blist->mark_deleted({
1229                         trans_id => $other_trans_id,
1230                     });
1231                 }
1232             }
1233         }
1234     }
1235
1236     #XXX Is this safe to do transactionally?
1237     # Free the place we're about to write to.
1238     if ( $blist->get_data_location_for({ allow_head => 0 }) ) {
1239         $blist->get_data_for({ allow_head => 0 })->free;
1240     }
1241
1242     $blist->write_md5({
1243         key      => $args->{key},
1244         key_md5  => $args->{key_md5},
1245         value    => $args->{value},
1246     });
1247 }
1248
1249 sub delete_key {
1250     my $self = shift;
1251     my ($args) = @_;
1252
1253     # XXX What should happen if this fails?
1254     my $blist = $self->get_bucket_list({
1255         key_md5 => $args->{key_md5},
1256     }) or die "How did delete_key fail (no blist)?!\n";
1257
1258     # Save the location so that we can free the data
1259     my $location = $blist->get_data_location_for({
1260         allow_head => 0,
1261     });
1262     my $old_value = $location && $self->engine->_load_sector( $location );
1263
1264     if ( $self->engine->trans_id == 0 ) {
1265         my @trans_ids = $self->engine->get_running_txn_ids;
1266         if ( @trans_ids ) {
1267             foreach my $other_trans_id ( @trans_ids ) {
1268                 next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1269                 $blist->write_md5({
1270                     trans_id => $other_trans_id,
1271                     key      => $args->{key},
1272                     key_md5  => $args->{key_md5},
1273                     value    => $old_value->clone,
1274                 });
1275             }
1276         }
1277     }
1278
1279     $blist->mark_deleted( $args );
1280
1281     my $data;
1282     if ( $old_value ) {
1283         $data = $old_value->data;
1284         $old_value->free;
1285     }
1286
1287     return $data;
1288 }
1289
1290 sub get_blist_loc {
1291     my $self = shift;
1292
1293     my $e = $self->engine;
1294     my $blist_loc = $e->storage->read_at( $self->offset + $self->base_size, $e->byte_size );
1295     return unpack( $StP{$e->byte_size}, $blist_loc );
1296 }
1297
1298 sub get_bucket_list {
1299     my $self = shift;
1300     my ($args) = @_;
1301     $args ||= {};
1302
1303     # XXX Add in check here for recycling?
1304
1305     my $engine = $self->engine;
1306
1307     my $blist_loc = $self->get_blist_loc;
1308
1309     # There's no index or blist yet
1310     unless ( $blist_loc ) {
1311         return unless $args->{create};
1312
1313         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1314             engine  => $engine,
1315             key_md5 => $args->{key_md5},
1316         });
1317
1318         $engine->storage->print_at( $self->offset + $self->base_size,
1319             pack( $StP{$engine->byte_size}, $blist->offset ),
1320         );
1321
1322         return $blist;
1323     }
1324
1325     # Add searching here through the index layers, if any
1326     my $sector = $engine->_load_sector( $blist_loc )
1327         or die "Cannot read sector at $blist_loc in get_bucket_list()";
1328     my $i = 0;
1329     my $last_sector = undef;
1330     while ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
1331         $blist_loc = $sector->get_entry( ord( substr( $args->{key_md5}, $i++, 1 ) ) );
1332         $last_sector = $sector;
1333         if ( $blist_loc ) {
1334             $sector = $engine->_load_sector( $blist_loc )
1335                 or die "Cannot read sector at $blist_loc in get_bucket_list()";
1336         }
1337         else {
1338             $sector = undef;
1339             last;
1340         }
1341     }
1342
1343     # This means we went through the Index sector(s) and found an empty slot
1344     unless ( $sector ) {
1345         return unless $args->{create};
1346
1347         die "No last_sector when attempting to build a new entry"
1348             unless $last_sector;
1349
1350         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1351             engine  => $engine,
1352             key_md5 => $args->{key_md5},
1353         });
1354
1355         $last_sector->set_entry( ord( substr( $args->{key_md5}, $i - 1, 1 ) ) => $blist->offset );
1356
1357         return $blist;
1358     }
1359
1360     $sector->find_md5( $args->{key_md5} );
1361
1362     # See whether or not we need to reindex the bucketlist
1363     if ( !$sector->has_md5 && $args->{create} && $sector->{idx} == -1 ) {
1364         my $new_index = DBM::Deep::Engine::Sector::Index->new({
1365             engine => $engine,
1366         });
1367
1368         my %blist_cache;
1369         #XXX q.v. the comments for this function.
1370         foreach my $entry ( $sector->chopped_up ) {
1371             my ($spot, $md5) = @{$entry};
1372             my $idx = ord( substr( $md5, $i, 1 ) );
1373
1374             # XXX This is inefficient
1375             my $blist = $blist_cache{$idx}
1376                 ||= DBM::Deep::Engine::Sector::BucketList->new({
1377                     engine => $engine,
1378                 });
1379
1380             $new_index->set_entry( $idx => $blist->offset );
1381
1382             my $new_spot = $blist->write_at_next_open( $md5 );
1383             $engine->reindex_entry( $spot => $new_spot );
1384         }
1385
1386         # Handle the new item separately.
1387         {
1388             my $idx = ord( substr( $args->{key_md5}, $i, 1 ) );
1389             my $blist = $blist_cache{$idx}
1390                 ||= DBM::Deep::Engine::Sector::BucketList->new({
1391                     engine => $engine,
1392                 });
1393
1394             $new_index->set_entry( $idx => $blist->offset );
1395
1396             #XXX THIS IS HACKY!
1397             $blist->find_md5( $args->{key_md5} );
1398             $blist->write_md5({
1399                 key     => $args->{key},
1400                 key_md5 => $args->{key_md5},
1401                 value   => DBM::Deep::Engine::Sector::Null->new({
1402                     engine => $engine,
1403                     data   => undef,
1404                 }),
1405             });
1406         }
1407
1408         if ( $last_sector ) {
1409             $last_sector->set_entry(
1410                 ord( substr( $args->{key_md5}, $i - 1, 1 ) ),
1411                 $new_index->offset,
1412             );
1413         } else {
1414             $engine->storage->print_at( $self->offset + $self->base_size,
1415                 pack( $StP{$engine->byte_size}, $new_index->offset ),
1416             );
1417         }
1418
1419         $sector->free;
1420
1421         $sector = $blist_cache{ ord( substr( $args->{key_md5}, $i, 1 ) ) };
1422         $sector->find_md5( $args->{key_md5} );
1423     }
1424
1425     return $sector;
1426 }
1427
1428 sub get_class_offset {
1429     my $self = shift;
1430
1431     my $e = $self->engine;
1432     return unpack(
1433         $StP{$e->byte_size},
1434         $e->storage->read_at(
1435             $self->offset + $self->base_size + 1 * $e->byte_size, $e->byte_size,
1436         ),
1437     );
1438 }
1439
1440 sub get_classname {
1441     my $self = shift;
1442
1443     my $class_offset = $self->get_class_offset;
1444
1445     return unless $class_offset;
1446
1447     return $self->engine->_load_sector( $class_offset )->data;
1448 }
1449
1450 sub data {
1451     my $self = shift;
1452
1453     my $new_obj = DBM::Deep->new({
1454         type        => $self->type,
1455         base_offset => $self->offset,
1456         staleness   => $self->staleness,
1457         storage     => $self->engine->storage,
1458         engine      => $self->engine,
1459     });
1460
1461     if ( $self->engine->storage->{autobless} ) {
1462         my $classname = $self->get_classname;
1463         if ( defined $classname ) {
1464             bless $new_obj, $classname;
1465         }
1466     }
1467
1468     return $new_obj;
1469 }
1470
1471 package DBM::Deep::Engine::Sector::BucketList;
1472
1473 our @ISA = qw( DBM::Deep::Engine::Sector );
1474
1475 sub _init {
1476     my $self = shift;
1477
1478     my $engine = $self->engine;
1479
1480     unless ( $self->offset ) {
1481         my $leftover = $self->size - $self->base_size;
1482
1483         $self->{offset} = $engine->_request_blist_sector( $self->size );
1484         $engine->storage->print_at( $self->offset, $engine->SIG_BLIST ); # Sector type
1485         # Skip staleness counter
1486         $engine->storage->print_at( $self->offset + $self->base_size,
1487             chr(0) x $leftover, # Zero-fill the data
1488         );
1489     }
1490
1491     if ( $self->{key_md5} ) {
1492         $self->find_md5;
1493     }
1494
1495     return $self;
1496 }
1497
1498 sub size {
1499     my $self = shift;
1500     unless ( $self->{size} ) {
1501         my $e = $self->engine;
1502         # Base + numbuckets * bucketsize
1503         $self->{size} = $self->base_size + $e->max_buckets * $self->bucket_size;
1504     }
1505     return $self->{size};
1506 }
1507
1508 sub free_meth { return '_add_free_blist_sector' }
1509
1510 sub bucket_size {
1511     my $self = shift;
1512     unless ( $self->{bucket_size} ) {
1513         my $e = $self->engine;
1514         # Key + head (location) + transactions (location + staleness-counter)
1515         my $location_size = $e->byte_size + $e->num_txns * ( $e->byte_size + 4 );
1516         $self->{bucket_size} = $e->hash_size + $location_size;
1517     }
1518     return $self->{bucket_size};
1519 }
1520
1521 # XXX This is such a poor hack. I need to rethink this code.
1522 sub chopped_up {
1523     my $self = shift;
1524
1525     my $e = $self->engine;
1526
1527     my @buckets;
1528     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1529         my $spot = $self->offset + $self->base_size + $idx * $self->bucket_size;
1530         my $md5 = $e->storage->read_at( $spot, $e->hash_size );
1531
1532         last if $md5 eq $e->blank_md5;
1533
1534         my $rest = $e->storage->read_at( undef, $self->bucket_size - $e->hash_size );
1535         push @buckets, [ $spot, $md5 . $rest ];
1536     }
1537
1538     return @buckets;
1539 }
1540
1541 sub write_at_next_open {
1542     my $self = shift;
1543     my ($entry) = @_;
1544
1545     #XXX This is such a hack!
1546     $self->{_next_open} = 0 unless exists $self->{_next_open};
1547
1548     my $spot = $self->offset + $self->base_size + $self->{_next_open}++ * $self->bucket_size;
1549     $self->engine->storage->print_at( $spot, $entry );
1550
1551     return $spot;
1552 }
1553
1554 sub has_md5 {
1555     my $self = shift;
1556     unless ( exists $self->{found} ) {
1557         $self->find_md5;
1558     }
1559     return $self->{found};
1560 }
1561
1562 sub find_md5 {
1563     my $self = shift;
1564
1565     $self->{found} = undef;
1566     $self->{idx}   = -1;
1567
1568     if ( @_ ) {
1569         $self->{key_md5} = shift;
1570     }
1571
1572     # If we don't have an MD5, then what are we supposed to do?
1573     unless ( exists $self->{key_md5} ) {
1574         DBM::Deep->_throw_error( "Cannot find_md5 without a key_md5 set" );
1575     }
1576
1577     my $e = $self->engine;
1578     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1579         my $potential = $e->storage->read_at(
1580             $self->offset + $self->base_size + $idx * $self->bucket_size, $e->hash_size,
1581         );
1582
1583         if ( $potential eq $e->blank_md5 ) {
1584             $self->{idx} = $idx;
1585             return;
1586         }
1587
1588         if ( $potential eq $self->{key_md5} ) {
1589             $self->{found} = 1;
1590             $self->{idx} = $idx;
1591             return;
1592         }
1593     }
1594
1595     return;
1596 }
1597
1598 sub write_md5 {
1599     my $self = shift;
1600     my ($args) = @_;
1601
1602     DBM::Deep->_throw_error( "write_md5: no key" ) unless exists $args->{key};
1603     DBM::Deep->_throw_error( "write_md5: no key_md5" ) unless exists $args->{key_md5};
1604     DBM::Deep->_throw_error( "write_md5: no value" ) unless exists $args->{value};
1605
1606     my $engine = $self->engine;
1607
1608     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
1609
1610     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1611     $engine->add_entry( $args->{trans_id}, $spot );
1612
1613     unless ($self->{found}) {
1614         my $key_sector = DBM::Deep::Engine::Sector::Scalar->new({
1615             engine => $engine,
1616             data   => $args->{key},
1617         });
1618
1619         $engine->storage->print_at( $spot,
1620             $args->{key_md5},
1621             pack( $StP{$engine->byte_size}, $key_sector->offset ),
1622         );
1623     }
1624
1625     my $loc = $spot
1626       + $engine->hash_size
1627       + $engine->byte_size
1628       + $args->{trans_id} * ( $engine->byte_size + 4 );
1629
1630     $engine->storage->print_at( $loc,
1631         pack( $StP{$engine->byte_size}, $args->{value}->offset ),
1632         pack( 'N', $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
1633     );
1634 }
1635
1636 sub mark_deleted {
1637     my $self = shift;
1638     my ($args) = @_;
1639     $args ||= {};
1640
1641     my $engine = $self->engine;
1642
1643     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
1644
1645     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1646     $engine->add_entry( $args->{trans_id}, $spot );
1647
1648     my $loc = $spot
1649       + $engine->hash_size
1650       + $engine->byte_size
1651       + $args->{trans_id} * ( $engine->byte_size + 4 );
1652
1653     $engine->storage->print_at( $loc,
1654         pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
1655         pack( 'N', $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
1656     );
1657 }
1658
1659 sub delete_md5 {
1660     my $self = shift;
1661     my ($args) = @_;
1662
1663     my $engine = $self->engine;
1664     return undef unless $self->{found};
1665
1666     # Save the location so that we can free the data
1667     my $location = $self->get_data_location_for({
1668         allow_head => 0,
1669     });
1670     my $key_sector = $self->get_key_for;
1671
1672     #XXX This isn't going to work right and you know it! This eradicates data
1673     # that we're not ready to eradicate just yet.
1674     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1675     $engine->storage->print_at( $spot,
1676         $engine->storage->read_at(
1677             $spot + $self->bucket_size,
1678             $self->bucket_size * ( $engine->num_txns - $self->{idx} - 1 ),
1679         ),
1680         chr(0) x $self->bucket_size,
1681     );
1682
1683     $key_sector->free;
1684
1685     my $data_sector = $self->engine->_load_sector( $location );
1686     my $data = $data_sector->data;
1687     $data_sector->free;
1688
1689     return $data;
1690 }
1691
1692 sub get_data_location_for {
1693     my $self = shift;
1694     my ($args) = @_;
1695     $args ||= {};
1696
1697     $args->{allow_head} = 0 unless exists $args->{allow_head};
1698     $args->{trans_id}   = $self->engine->trans_id unless exists $args->{trans_id};
1699     $args->{idx}        = $self->{idx} unless exists $args->{idx};
1700
1701     my $e = $self->engine;
1702
1703     my $spot = $self->offset + $self->base_size
1704       + $args->{idx} * $self->bucket_size
1705       + $e->hash_size
1706       + $e->byte_size
1707       + $args->{trans_id} * ( $e->byte_size + 4 );
1708
1709     my $buffer = $e->storage->read_at(
1710         $spot,
1711         $e->byte_size + 4,
1712     );
1713     my ($loc, $staleness) = unpack( $StP{$e->byte_size} . ' N', $buffer );
1714
1715     # We have found an entry that is old, so get rid of it
1716     if ( $staleness != (my $s = $e->get_txn_staleness_counter( $args->{trans_id} ) ) ) {
1717         $e->storage->print_at(
1718             $spot,
1719             pack( $StP{$e->byte_size} . ' N', (0) x 2 ),
1720         );
1721         $loc = 0;
1722     }
1723
1724     # If we're in a transaction and we never wrote to this location, try the
1725     # HEAD instead.
1726     if ( $args->{trans_id} && !$loc && $args->{allow_head} ) {
1727         return $self->get_data_location_for({
1728             trans_id   => 0,
1729             allow_head => 1,
1730             idx        => $args->{idx},
1731         });
1732     }
1733     return $loc <= 1 ? 0 : $loc;
1734 }
1735
1736 sub get_data_for {
1737     my $self = shift;
1738     my ($args) = @_;
1739     $args ||= {};
1740
1741     return unless $self->{found};
1742     my $location = $self->get_data_location_for({
1743         allow_head => $args->{allow_head},
1744     });
1745     return $self->engine->_load_sector( $location );
1746 }
1747
1748 sub get_key_for {
1749     my $self = shift;
1750     my ($idx) = @_;
1751     $idx = $self->{idx} unless defined $idx;
1752
1753     if ( $idx >= $self->engine->max_buckets ) {
1754         DBM::Deep->_throw_error( "get_key_for(): Attempting to retrieve $idx" );
1755     }
1756
1757     my $location = $self->engine->storage->read_at(
1758         $self->offset + $self->base_size + $idx * $self->bucket_size + $self->engine->hash_size,
1759         $self->engine->byte_size,
1760     );
1761     $location = unpack( $StP{$self->engine->byte_size}, $location );
1762     return unless $location;
1763     return $self->engine->_load_sector( $location );
1764 }
1765
1766 package DBM::Deep::Engine::Sector::Index;
1767
1768 our @ISA = qw( DBM::Deep::Engine::Sector );
1769
1770 sub _init {
1771     my $self = shift;
1772
1773     my $engine = $self->engine;
1774
1775     unless ( $self->offset ) {
1776         my $leftover = $self->size - $self->base_size;
1777
1778         $self->{offset} = $engine->_request_index_sector( $self->size );
1779         $engine->storage->print_at( $self->offset, $engine->SIG_INDEX ); # Sector type
1780         # Skip staleness counter
1781         $engine->storage->print_at( $self->offset + $self->base_size,
1782             chr(0) x $leftover, # Zero-fill the rest
1783         );
1784     }
1785
1786     return $self;
1787 }
1788
1789 sub size {
1790     my $self = shift;
1791     unless ( $self->{size} ) {
1792         my $e = $self->engine;
1793         $self->{size} = $self->base_size + $e->byte_size * $e->hash_chars;
1794     }
1795     return $self->{size};
1796 }
1797
1798 sub free_meth { return '_add_free_index_sector' }
1799
1800 sub free {
1801     my $self = shift;
1802     my $e = $self->engine;
1803
1804     for my $i ( 0 .. $e->hash_chars - 1 ) {
1805         my $l = $self->location_for( $i ) or next;
1806         $e->_load_sector( $l )->free;
1807     }
1808
1809     $self->SUPER::free();
1810 }
1811
1812 sub _loc_for {
1813     my $self = shift;
1814     my ($idx) = @_;
1815     return $self->offset + $self->base_size + $idx * $self->engine->byte_size;
1816 }
1817
1818 sub get_entry {
1819     my $self = shift;
1820     my ($idx) = @_;
1821
1822     my $e = $self->engine;
1823
1824     die "get_entry: Out of range ($idx)"
1825         if $idx < 0 || $idx >= $e->hash_chars;
1826
1827     return unpack(
1828         $StP{$e->byte_size},
1829         $e->storage->read_at( $self->_loc_for( $idx ), $e->byte_size ),
1830     );
1831 }
1832
1833 sub set_entry {
1834     my $self = shift;
1835     my ($idx, $loc) = @_;
1836
1837     my $e = $self->engine;
1838
1839     die "set_entry: Out of range ($idx)"
1840         if $idx < 0 || $idx >= $e->hash_chars;
1841
1842     $self->engine->storage->print_at(
1843         $self->_loc_for( $idx ),
1844         pack( $StP{$e->byte_size}, $loc ),
1845     );
1846 }
1847
1848 1;
1849 __END__