lib/DBM/Deep/Engine.pm

   1 package DBM::Deep::Engine;
   2
   3 use 5.006_000;
   4
   5 use strict;
   6 use warnings;
   7
   8 our $VERSION = q(1.0009);
   9
  10 use Scalar::Util ();
  11
  12 # File-wide notes:
  13 # * Every method in here assumes that the storage has been appropriately
  14 #   safeguarded. This can be anything from flock() to some sort of manual
  15 #   mutex. But, it's the caller's responsability to make sure that this has
  16 #   been done.
  17
  18 # Setup file and tag signatures.  These should never change.
  19 sub SIG_FILE     () { 'DPDB' }
  20 sub SIG_HEADER   () { 'h'    }
  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_SIZE     () {  1     }
  29
  30 my $STALE_SIZE = 2;
  31
  32 # Please refer to the pack() documentation for further information
  33 my %StP = (
  34     1 => 'C', # Unsigned char value (no order needed as it's just one byte)
  35     2 => 'n', # Unsigned short in "network" (big-endian) order
  36     4 => 'N', # Unsigned long in "network" (big-endian) order
  37     8 => 'Q', # Usigned quad (no order specified, presumably machine-dependent)
  38 );
  39
  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    => 1,   # The HEAD
  54         trans_id    => 0,   # Default to the HEAD
  55
  56         data_sector_size => 64, # Size in bytes of each data sector
  57
  58         entries => {}, # This is the list of entries for transactions
  59         storage => undef,
  60     }, $class;
  61
  62     # Never allow byte_size to be set directly.
  63     delete $args->{byte_size};
  64     if ( defined $args->{pack_size} ) {
  65         if ( lc $args->{pack_size} eq 'small' ) {
  66             $args->{byte_size} = 2;
  67         }
  68         elsif ( lc $args->{pack_size} eq 'medium' ) {
  69             $args->{byte_size} = 4;
  70         }
  71         elsif ( lc $args->{pack_size} eq 'large' ) {
  72             $args->{byte_size} = 8;
  73         }
  74         else {
  75             DBM::Deep->_throw_error( "Unknown pack_size value: '$args->{pack_size}'" );
  76         }
  77     }
  78
  79     # Grab the parameters we want to use
  80     foreach my $param ( keys %$self ) {
  81         next unless exists $args->{$param};
  82         $self->{$param} = $args->{$param};
  83     }
  84
  85     my %validations = (
  86         max_buckets      => { floor => 16, ceil => 256 },
  87         num_txns         => { floor => 1,  ceil => 255 },
  88         data_sector_size => { floor => 32, ceil => 256 },
  89     );
  90
  91     while ( my ($attr, $c) = each %validations ) {
  92         if (   !defined $self->{$attr}
  93             || !length $self->{$attr}
  94             || $self->{$attr} =~ /\D/
  95             || $self->{$attr} < $c->{floor}
  96         ) {
  97             $self->{$attr} = '(undef)' if !defined $self->{$attr};
  98             warn "Floor of $attr is $c->{floor}. Setting it to $c->{floor} from '$self->{$attr}'\n";
  99             $self->{$attr} = $c->{floor};
 100         }
 101         elsif ( $self->{$attr} > $c->{ceil} ) {
 102             warn "Ceiling of $attr is $c->{ceil}. Setting it to $c->{ceil} from '$self->{$attr}'\n";
 103             $self->{$attr} = $c->{ceil};
 104         }
 105     }
 106
 107     if ( !$self->{digest} ) {
 108         require Digest::MD5;
 109         $self->{digest} = \&Digest::MD5::md5;
 110     }
 111
 112     return $self;
 113 }
 114
 115 ################################################################################
 116
 117 sub read_value {
 118     my $self = shift;
 119     my ($obj, $key) = @_;
 120
 121     # This will be a Reference sector
 122     my $sector = $self->_load_sector( $obj->_base_offset )
 123         or return;
 124
 125     if ( $sector->staleness != $obj->_staleness ) {
 126         return;
 127     }
 128
 129     my $key_md5 = $self->_apply_digest( $key );
 130
 131     my $value_sector = $sector->get_data_for({
 132         key_md5    => $key_md5,
 133         allow_head => 1,
 134     });
 135
 136     unless ( $value_sector ) {
 137         $value_sector = DBM::Deep::Engine::Sector::Null->new({
 138             engine => $self,
 139             data   => undef,
 140         });
 141
 142         $sector->write_data({
 143             key_md5 => $key_md5,
 144             key     => $key,
 145             value   => $value_sector,
 146         });
 147     }
 148
 149     return $value_sector->data;
 150 }
 151
 152 sub get_classname {
 153     my $self = shift;
 154     my ($obj) = @_;
 155
 156     # This will be a Reference sector
 157     my $sector = $self->_load_sector( $obj->_base_offset )
 158         or DBM::Deep->_throw_error( "How did get_classname fail (no sector for '$obj')?!" );
 159
 160     if ( $sector->staleness != $obj->_staleness ) {
 161         return;
 162     }
 163
 164     return $sector->get_classname;
 165 }
 166
 167 sub make_reference {
 168     my $self = shift;
 169     my ($obj, $old_key, $new_key) = @_;
 170
 171     # This will be a Reference sector
 172     my $sector = $self->_load_sector( $obj->_base_offset )
 173         or DBM::Deep->_throw_error( "How did get_classname fail (no sector for '$obj')?!" );
 174
 175     if ( $sector->staleness != $obj->_staleness ) {
 176         return;
 177     }
 178
 179     my $old_md5 = $self->_apply_digest( $old_key );
 180
 181     my $value_sector = $sector->get_data_for({
 182         key_md5    => $old_md5,
 183         allow_head => 1,
 184     });
 185
 186     unless ( $value_sector ) {
 187         $value_sector = DBM::Deep::Engine::Sector::Null->new({
 188             engine => $self,
 189             data   => undef,
 190         });
 191
 192         $sector->write_data({
 193             key_md5 => $old_md5,
 194             key     => $old_key,
 195             value   => $value_sector,
 196         });
 197     }
 198
 199     if ( $value_sector->isa( 'DBM::Deep::Engine::Sector::Reference' ) ) {
 200         $sector->write_data({
 201             key     => $new_key,
 202             key_md5 => $self->_apply_digest( $new_key ),
 203             value   => $value_sector,
 204         });
 205         $value_sector->increment_refcount;
 206     }
 207     else {
 208         $sector->write_data({
 209             key     => $new_key,
 210             key_md5 => $self->_apply_digest( $new_key ),
 211             value   => $value_sector->clone,
 212         });
 213     }
 214 }
 215
 216 sub key_exists {
 217     my $self = shift;
 218     my ($obj, $key) = @_;
 219
 220     # This will be a Reference sector
 221     my $sector = $self->_load_sector( $obj->_base_offset )
 222         or return '';
 223
 224     if ( $sector->staleness != $obj->_staleness ) {
 225         return '';
 226     }
 227
 228     my $data = $sector->get_data_for({
 229         key_md5    => $self->_apply_digest( $key ),
 230         allow_head => 1,
 231     });
 232
 233     # exists() returns 1 or '' for true/false.
 234     return $data ? 1 : '';
 235 }
 236
 237 sub delete_key {
 238     my $self = shift;
 239     my ($obj, $key) = @_;
 240
 241     my $sector = $self->_load_sector( $obj->_base_offset )
 242         or return;
 243
 244     if ( $sector->staleness != $obj->_staleness ) {
 245         return;
 246     }
 247
 248     return $sector->delete_key({
 249         key_md5    => $self->_apply_digest( $key ),
 250         allow_head => 0,
 251     });
 252 }
 253
 254 sub write_value {
 255     my $self = shift;
 256     my ($obj, $key, $value) = @_;
 257
 258     my $r = Scalar::Util::reftype( $value ) || '';
 259     {
 260         last if $r eq '';
 261         last if $r eq 'HASH';
 262         last if $r eq 'ARRAY';
 263
 264         DBM::Deep->_throw_error(
 265             "Storage of references of type '$r' is not supported."
 266         );
 267     }
 268
 269     # This will be a Reference sector
 270     my $sector = $self->_load_sector( $obj->_base_offset )
 271         or DBM::Deep->_throw_error( "Cannot write to a deleted spot in DBM::Deep." );
 272
 273     if ( $sector->staleness != $obj->_staleness ) {
 274         DBM::Deep->_throw_error( "Cannot write to a deleted spot in DBM::Deep.n" );
 275     }
 276
 277     my ($class, $type);
 278     if ( !defined $value ) {
 279         $class = 'DBM::Deep::Engine::Sector::Null';
 280     }
 281     elsif ( $r eq 'ARRAY' || $r eq 'HASH' ) {
 282         my $is_dbm_deep = eval { local $SIG{'__DIE__'}; $value->isa( 'DBM::Deep' ); };
 283         if ( $is_dbm_deep ) {
 284             if ( $value->_engine->storage == $self->storage ) {
 285                 my $value_sector = $self->_load_sector( $value->_base_offset );
 286                 $sector->write_data({
 287                     key     => $key,
 288                     key_md5 => $self->_apply_digest( $key ),
 289                     value   => $value_sector,
 290                 });
 291                 $value_sector->increment_refcount;
 292                 return 1;
 293             }
 294
 295             DBM::Deep->_throw_error( "Cannot store values across DBM::Deep files. Please use export() instead." );
 296         }
 297         if ( $r eq 'ARRAY' && tied(@$value) ) {
 298             DBM::Deep->_throw_error( "Cannot store something that is tied." );
 299         }
 300         if ( $r eq 'HASH' && tied(%$value) ) {
 301             DBM::Deep->_throw_error( "Cannot store something that is tied." );
 302         }
 303         $class = 'DBM::Deep::Engine::Sector::Reference';
 304         $type = substr( $r, 0, 1 );
 305     }
 306     else {
 307         if ( tied($value) ) {
 308             DBM::Deep->_throw_error( "Cannot store something that is tied." );
 309         }
 310         $class = 'DBM::Deep::Engine::Sector::Scalar';
 311     }
 312
 313     # Create this after loading the reference sector in case something bad happens.
 314     # This way, we won't allocate value sector(s) needlessly.
 315     my $value_sector = $class->new({
 316         engine => $self,
 317         data   => $value,
 318         type   => $type,
 319     });
 320
 321     $sector->write_data({
 322         key     => $key,
 323         key_md5 => $self->_apply_digest( $key ),
 324         value   => $value_sector,
 325     });
 326
 327     # This code is to make sure we write all the values in the $value to the disk
 328     # and to make sure all changes to $value after the assignment are reflected
 329     # on disk. This may be counter-intuitive at first, but it is correct dwimmery.
 330     #   NOTE - simply tying $value won't perform a STORE on each value. Hence, the
 331     # copy to a temp value.
 332     if ( $r eq 'ARRAY' ) {
 333         my @temp = @$value;
 334         tie @$value, 'DBM::Deep', {
 335             base_offset => $value_sector->offset,
 336             staleness   => $value_sector->staleness,
 337             storage     => $self->storage,
 338             engine      => $self,
 339         };
 340         @$value = @temp;
 341         bless $value, 'DBM::Deep::Array' unless Scalar::Util::blessed( $value );
 342     }
 343     elsif ( $r eq 'HASH' ) {
 344         my %temp = %$value;
 345         tie %$value, 'DBM::Deep', {
 346             base_offset => $value_sector->offset,
 347             staleness   => $value_sector->staleness,
 348             storage     => $self->storage,
 349             engine      => $self,
 350         };
 351
 352         %$value = %temp;
 353         bless $value, 'DBM::Deep::Hash' unless Scalar::Util::blessed( $value );
 354     }
 355
 356     return 1;
 357 }
 358
 359 # XXX Add staleness here
 360 sub get_next_key {
 361     my $self = shift;
 362     my ($obj, $prev_key) = @_;
 363
 364     # XXX Need to add logic about resetting the iterator if any key in the reference has changed
 365     unless ( $prev_key ) {
 366         $obj->{iterator} = DBM::Deep::Iterator->new({
 367             base_offset => $obj->_base_offset,
 368             engine      => $self,
 369         });
 370     }
 371
 372     return $obj->{iterator}->get_next_key( $obj );
 373 }
 374
 375 ################################################################################
 376
 377 sub setup_fh {
 378     my $self = shift;
 379     my ($obj) = @_;
 380
 381     # We're opening the file.
 382     unless ( $obj->_base_offset ) {
 383         my $bytes_read = $self->_read_file_header;
 384
 385         # Creating a new file
 386         unless ( $bytes_read ) {
 387             $self->_write_file_header;
 388
 389             # 1) Create Array/Hash entry
 390             my $initial_reference = DBM::Deep::Engine::Sector::Reference->new({
 391                 engine => $self,
 392                 type   => $obj->_type,
 393             });
 394             $obj->{base_offset} = $initial_reference->offset;
 395             $obj->{staleness} = $initial_reference->staleness;
 396
 397             $self->storage->flush;
 398         }
 399         # Reading from an existing file
 400         else {
 401             $obj->{base_offset} = $bytes_read;
 402             my $initial_reference = DBM::Deep::Engine::Sector::Reference->new({
 403                 engine => $self,
 404                 offset => $obj->_base_offset,
 405             });
 406             unless ( $initial_reference ) {
 407                 DBM::Deep->_throw_error("Corrupted file, no master index record");
 408             }
 409
 410             unless ($obj->_type eq $initial_reference->type) {
 411                 DBM::Deep->_throw_error("File type mismatch");
 412             }
 413
 414             $obj->{staleness} = $initial_reference->staleness;
 415         }
 416     }
 417
 418     return 1;
 419 }
 420
 421 sub begin_work {
 422     my $self = shift;
 423     my ($obj) = @_;
 424
 425     if ( $self->trans_id ) {
 426         DBM::Deep->_throw_error( "Cannot begin_work within an active transaction" );
 427     }
 428
 429     my @slots = $self->read_txn_slots;
 430     my $found;
 431     for my $i ( 0 .. $#slots ) {
 432         next if $slots[$i];
 433
 434         $slots[$i] = 1;
 435         $self->set_trans_id( $i + 1 );
 436         $found = 1;
 437         last;
 438     }
 439     unless ( $found ) {
 440         DBM::Deep->_throw_error( "Cannot allocate transaction ID" );
 441     }
 442     $self->write_txn_slots( @slots );
 443
 444     if ( !$self->trans_id ) {
 445         DBM::Deep->_throw_error( "Cannot begin_work - no available transactions" );
 446     }
 447
 448     return;
 449 }
 450
 451 sub rollback {
 452     my $self = shift;
 453     my ($obj) = @_;
 454
 455     if ( !$self->trans_id ) {
 456         DBM::Deep->_throw_error( "Cannot rollback without an active transaction" );
 457     }
 458
 459     # Each entry is the file location for a bucket that has a modification for
 460     # this transaction. The entries need to be expunged.
 461     foreach my $entry (@{ $self->get_entries } ) {
 462         # Remove the entry here
 463         my $read_loc = $entry
 464           + $self->hash_size
 465           + $self->byte_size
 466           + $self->byte_size
 467           + ($self->trans_id - 1) * ( $self->byte_size + $STALE_SIZE );
 468
 469         my $data_loc = $self->storage->read_at( $read_loc, $self->byte_size );
 470         $data_loc = unpack( $StP{$self->byte_size}, $data_loc );
 471         $self->storage->print_at( $read_loc, pack( $StP{$self->byte_size}, 0 ) );
 472
 473         if ( $data_loc > 1 ) {
 474             $self->_load_sector( $data_loc )->free;
 475         }
 476     }
 477
 478     $self->clear_entries;
 479
 480     my @slots = $self->read_txn_slots;
 481     $slots[$self->trans_id-1] = 0;
 482     $self->write_txn_slots( @slots );
 483     $self->inc_txn_staleness_counter( $self->trans_id );
 484     $self->set_trans_id( 0 );
 485
 486     return 1;
 487 }
 488
 489 sub commit {
 490     my $self = shift;
 491     my ($obj) = @_;
 492
 493     if ( !$self->trans_id ) {
 494         DBM::Deep->_throw_error( "Cannot commit without an active transaction" );
 495     }
 496
 497     foreach my $entry (@{ $self->get_entries } ) {
 498         # Overwrite the entry in head with the entry in trans_id
 499         my $base = $entry
 500           + $self->hash_size
 501           + $self->byte_size;
 502
 503         my $head_loc = $self->storage->read_at( $base, $self->byte_size );
 504         $head_loc = unpack( $StP{$self->byte_size}, $head_loc );
 505
 506         my $spot = $base + $self->byte_size + ($self->trans_id - 1) * ( $self->byte_size + $STALE_SIZE );
 507         my $trans_loc = $self->storage->read_at(
 508             $spot, $self->byte_size,
 509         );
 510
 511         $self->storage->print_at( $base, $trans_loc );
 512         $self->storage->print_at(
 513             $spot,
 514             pack( $StP{$self->byte_size} . ' ' . $StP{$STALE_SIZE}, (0) x 2 ),
 515         );
 516
 517         if ( $head_loc > 1 ) {
 518             $self->_load_sector( $head_loc )->free;
 519         }
 520     }
 521
 522     $self->clear_entries;
 523
 524     my @slots = $self->read_txn_slots;
 525     $slots[$self->trans_id-1] = 0;
 526     $self->write_txn_slots( @slots );
 527     $self->inc_txn_staleness_counter( $self->trans_id );
 528     $self->set_trans_id( 0 );
 529
 530     return 1;
 531 }
 532
 533 sub read_txn_slots {
 534     my $self = shift;
 535     my $bl = $self->txn_bitfield_len;
 536     my $num_bits = $bl * 8;
 537     return split '', unpack( 'b'.$num_bits,
 538         $self->storage->read_at(
 539             $self->trans_loc, $bl,
 540         )
 541     );
 542 }
 543
 544 sub write_txn_slots {
 545     my $self = shift;
 546     my $num_bits = $self->txn_bitfield_len * 8;
 547     $self->storage->print_at( $self->trans_loc,
 548         pack( 'b'.$num_bits, join('', @_) ),
 549     );
 550 }
 551
 552 sub get_running_txn_ids {
 553     my $self = shift;
 554     my @transactions = $self->read_txn_slots;
 555     my @trans_ids = map { $_+1} grep { $transactions[$_] } 0 .. $#transactions;
 556 }
 557
 558 sub get_txn_staleness_counter {
 559     my $self = shift;
 560     my ($trans_id) = @_;
 561
 562     # Hardcode staleness of 0 for the HEAD
 563     return 0 unless $trans_id;
 564
 565     return unpack( $StP{$STALE_SIZE},
 566         $self->storage->read_at(
 567             $self->trans_loc + $self->txn_bitfield_len + $STALE_SIZE * ($trans_id - 1),
 568             $STALE_SIZE,
 569         )
 570     );
 571 }
 572
 573 sub inc_txn_staleness_counter {
 574     my $self = shift;
 575     my ($trans_id) = @_;
 576
 577     # Hardcode staleness of 0 for the HEAD
 578     return 0 unless $trans_id;
 579
 580     $self->storage->print_at(
 581         $self->trans_loc + $self->txn_bitfield_len + $STALE_SIZE * ($trans_id - 1),
 582         pack( $StP{$STALE_SIZE}, $self->get_txn_staleness_counter( $trans_id ) + 1 ),
 583     );
 584 }
 585
 586 sub get_entries {
 587     my $self = shift;
 588     return [ keys %{ $self->{entries}{$self->trans_id} ||= {} } ];
 589 }
 590
 591 sub add_entry {
 592     my $self = shift;
 593     my ($trans_id, $loc) = @_;
 594
 595     $self->{entries}{$trans_id} ||= {};
 596     $self->{entries}{$trans_id}{$loc} = undef;
 597 }
 598
 599 # If the buckets are being relocated because of a reindexing, the entries
 600 # mechanism needs to be made aware of it.
 601 sub reindex_entry {
 602     my $self = shift;
 603     my ($old_loc, $new_loc) = @_;
 604
 605     TRANS:
 606     while ( my ($trans_id, $locs) = each %{ $self->{entries} } ) {
 607         foreach my $orig_loc ( keys %{ $locs } ) {
 608             if ( $orig_loc == $old_loc ) {
 609                 delete $locs->{orig_loc};
 610                 $locs->{$new_loc} = undef;
 611                 next TRANS;
 612             }
 613         }
 614     }
 615 }
 616
 617 sub clear_entries {
 618     my $self = shift;
 619     delete $self->{entries}{$self->trans_id};
 620 }
 621
 622 ################################################################################
 623
 624 {
 625     my $header_fixed = length( SIG_FILE ) + 1 + 4 + 4;
 626     my $this_file_version = 3;
 627
 628     sub _write_file_header {
 629         my $self = shift;
 630
 631         my $nt = $self->num_txns;
 632         my $bl = $self->txn_bitfield_len;
 633
 634         my $header_var = 1 + 1 + 1 + 1 + $bl + $STALE_SIZE * ($nt - 1) + 3 * $self->byte_size;
 635
 636         my $loc = $self->storage->request_space( $header_fixed + $header_var );
 637
 638         $self->storage->print_at( $loc,
 639             SIG_FILE,
 640             SIG_HEADER,
 641             pack('N', $this_file_version), # At this point, we're at 9 bytes
 642             pack('N', $header_var),        # header size
 643             # --- Above is $header_fixed. Below is $header_var
 644             pack('C', $self->byte_size),
 645
 646             # These shenanigans are to allow a 256 within a C
 647             pack('C', $self->max_buckets - 1),
 648             pack('C', $self->data_sector_size - 1),
 649
 650             pack('C', $nt),
 651             pack('C' . $bl, 0 ),                           # Transaction activeness bitfield
 652             pack($StP{$STALE_SIZE}.($nt-1), 0 x ($nt-1) ), # Transaction staleness counters
 653             pack($StP{$self->byte_size}, 0), # Start of free chain (blist size)
 654             pack($StP{$self->byte_size}, 0), # Start of free chain (data size)
 655             pack($StP{$self->byte_size}, 0), # Start of free chain (index size)
 656         );
 657
 658         #XXX Set these less fragilely
 659         $self->set_trans_loc( $header_fixed + 4 );
 660         $self->set_chains_loc( $header_fixed + 4 + $bl + $STALE_SIZE * ($nt-1) );
 661
 662         return;
 663     }
 664
 665     sub _read_file_header {
 666         my $self = shift;
 667
 668         my $buffer = $self->storage->read_at( 0, $header_fixed );
 669         return unless length($buffer);
 670
 671         my ($file_signature, $sig_header, $file_version, $size) = unpack(
 672             'A4 A N N', $buffer
 673         );
 674
 675         unless ( $file_signature eq SIG_FILE ) {
 676             $self->storage->close;
 677             DBM::Deep->_throw_error( "Signature not found -- file is not a Deep DB" );
 678         }
 679
 680         unless ( $sig_header eq SIG_HEADER ) {
 681             $self->storage->close;
 682             DBM::Deep->_throw_error( "Pre-1.00 file version found" );
 683         }
 684
 685         unless ( $file_version == $this_file_version ) {
 686             $self->storage->close;
 687             DBM::Deep->_throw_error(
 688                 "Wrong file version found - " .  $file_version .
 689                 " - expected " . $this_file_version
 690             );
 691         }
 692
 693         my $buffer2 = $self->storage->read_at( undef, $size );
 694         my @values = unpack( 'C C C C', $buffer2 );
 695
 696         if ( @values != 4 || grep { !defined } @values ) {
 697             $self->storage->close;
 698             DBM::Deep->_throw_error("Corrupted file - bad header");
 699         }
 700
 701         #XXX Add warnings if values weren't set right
 702         @{$self}{qw(byte_size max_buckets data_sector_size num_txns)} = @values;
 703
 704         # These shenangians are to allow a 256 within a C
 705         $self->{max_buckets} += 1;
 706         $self->{data_sector_size} += 1;
 707
 708         my $bl = $self->txn_bitfield_len;
 709
 710         my $header_var = scalar(@values) + $bl + $STALE_SIZE * ($self->num_txns - 1) + 3 * $self->byte_size;
 711         unless ( $size == $header_var ) {
 712             $self->storage->close;
 713             DBM::Deep->_throw_error( "Unexpected size found ($size <-> $header_var)." );
 714         }
 715
 716         $self->set_trans_loc( $header_fixed + scalar(@values) );
 717         $self->set_chains_loc( $header_fixed + scalar(@values) + $bl + $STALE_SIZE * ($self->num_txns - 1) );
 718
 719         return length($buffer) + length($buffer2);
 720     }
 721 }
 722
 723 sub _load_sector {
 724     my $self = shift;
 725     my ($offset) = @_;
 726
 727     # Add a catch for offset of 0 or 1
 728     return if !$offset || $offset <= 1;
 729
 730     my $type = $self->storage->read_at( $offset, 1 );
 731     return if $type eq chr(0);
 732
 733     if ( $type eq $self->SIG_ARRAY || $type eq $self->SIG_HASH ) {
 734         return DBM::Deep::Engine::Sector::Reference->new({
 735             engine => $self,
 736             type   => $type,
 737             offset => $offset,
 738         });
 739     }
 740     # XXX Don't we need key_md5 here?
 741     elsif ( $type eq $self->SIG_BLIST ) {
 742         return DBM::Deep::Engine::Sector::BucketList->new({
 743             engine => $self,
 744             type   => $type,
 745             offset => $offset,
 746         });
 747     }
 748     elsif ( $type eq $self->SIG_INDEX ) {
 749         return DBM::Deep::Engine::Sector::Index->new({
 750             engine => $self,
 751             type   => $type,
 752             offset => $offset,
 753         });
 754     }
 755     elsif ( $type eq $self->SIG_NULL ) {
 756         return DBM::Deep::Engine::Sector::Null->new({
 757             engine => $self,
 758             type   => $type,
 759             offset => $offset,
 760         });
 761     }
 762     elsif ( $type eq $self->SIG_DATA ) {
 763         return DBM::Deep::Engine::Sector::Scalar->new({
 764             engine => $self,
 765             type   => $type,
 766             offset => $offset,
 767         });
 768     }
 769     # This was deleted from under us, so just return and let the caller figure it out.
 770     elsif ( $type eq $self->SIG_FREE ) {
 771         return;
 772     }
 773
 774     DBM::Deep->_throw_error( "'$offset': Don't know what to do with type '$type'" );
 775 }
 776
 777 sub _apply_digest {
 778     my $self = shift;
 779     return $self->{digest}->(@_);
 780 }
 781
 782 sub _add_free_blist_sector { shift->_add_free_sector( 0, @_ ) }
 783 sub _add_free_data_sector { shift->_add_free_sector( 1, @_ ) }
 784 sub _add_free_index_sector { shift->_add_free_sector( 2, @_ ) }
 785
 786 sub _add_free_sector {
 787     my $self = shift;
 788     my ($multiple, $offset, $size) = @_;
 789
 790     my $chains_offset = $multiple * $self->byte_size;
 791
 792     my $storage = $self->storage;
 793
 794     # Increment staleness.
 795     # XXX Can this increment+modulo be done by "&= 0x1" ?
 796     my $staleness = unpack( $StP{$STALE_SIZE}, $storage->read_at( $offset + SIG_SIZE, $STALE_SIZE ) );
 797     $staleness = ($staleness + 1 ) % ( 2 ** ( 8 * $STALE_SIZE ) );
 798     $storage->print_at( $offset + SIG_SIZE, pack( $StP{$STALE_SIZE}, $staleness ) );
 799
 800     my $old_head = $storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 801
 802     $storage->print_at( $self->chains_loc + $chains_offset,
 803         pack( $StP{$self->byte_size}, $offset ),
 804     );
 805
 806     # Record the old head in the new sector after the signature and staleness counter
 807     $storage->print_at( $offset + SIG_SIZE + $STALE_SIZE, $old_head );
 808 }
 809
 810 sub _request_blist_sector { shift->_request_sector( 0, @_ ) }
 811 sub _request_data_sector { shift->_request_sector( 1, @_ ) }
 812 sub _request_index_sector { shift->_request_sector( 2, @_ ) }
 813
 814 sub _request_sector {
 815     my $self = shift;
 816     my ($multiple, $size) = @_;
 817
 818     my $chains_offset = $multiple * $self->byte_size;
 819
 820     my $old_head = $self->storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 821     my $loc = unpack( $StP{$self->byte_size}, $old_head );
 822
 823     # We don't have any free sectors of the right size, so allocate a new one.
 824     unless ( $loc ) {
 825         my $offset = $self->storage->request_space( $size );
 826
 827         # Zero out the new sector. This also guarantees correct increases
 828         # in the filesize.
 829         $self->storage->print_at( $offset, chr(0) x $size );
 830
 831         return $offset;
 832     }
 833
 834     # Read the new head after the signature and the staleness counter
 835     my $new_head = $self->storage->read_at( $loc + SIG_SIZE + $STALE_SIZE, $self->byte_size );
 836     $self->storage->print_at( $self->chains_loc + $chains_offset, $new_head );
 837     $self->storage->print_at(
 838         $loc + SIG_SIZE + $STALE_SIZE,
 839         pack( $StP{$self->byte_size}, 0 ),
 840     );
 841
 842     return $loc;
 843 }
 844
 845 ################################################################################
 846
 847 sub storage     { $_[0]{storage} }
 848 sub byte_size   { $_[0]{byte_size} }
 849 sub hash_size   { $_[0]{hash_size} }
 850 sub hash_chars  { $_[0]{hash_chars} }
 851 sub num_txns    { $_[0]{num_txns} }
 852 sub max_buckets { $_[0]{max_buckets} }
 853 sub blank_md5   { chr(0) x $_[0]->hash_size }
 854 sub data_sector_size { $_[0]{data_sector_size} }
 855
 856 # This is a calculated value
 857 sub txn_bitfield_len {
 858     my $self = shift;
 859     unless ( exists $self->{txn_bitfield_len} ) {
 860         my $temp = ($self->num_txns) / 8;
 861         if ( $temp > int( $temp ) ) {
 862             $temp = int( $temp ) + 1;
 863         }
 864         $self->{txn_bitfield_len} = $temp;
 865     }
 866     return $self->{txn_bitfield_len};
 867 }
 868
 869 sub trans_id     { $_[0]{trans_id} }
 870 sub set_trans_id { $_[0]{trans_id} = $_[1] }
 871
 872 sub trans_loc     { $_[0]{trans_loc} }
 873 sub set_trans_loc { $_[0]{trans_loc} = $_[1] }
 874
 875 sub chains_loc     { $_[0]{chains_loc} }
 876 sub set_chains_loc { $_[0]{chains_loc} = $_[1] }
 877
 878 sub cache       { $_[0]{cache} ||= {} }
 879 sub clear_cache { %{$_[0]->cache} = () }
 880
 881 sub _dump_file {
 882     my $self = shift;
 883
 884     # Read the header
 885     my $spot = $self->_read_file_header();
 886
 887     my %types = (
 888         0 => 'B',
 889         1 => 'D',
 890         2 => 'I',
 891     );
 892
 893     my %sizes = (
 894         'D' => $self->data_sector_size,
 895         'B' => DBM::Deep::Engine::Sector::BucketList->new({engine=>$self,offset=>1})->size,
 896         'I' => DBM::Deep::Engine::Sector::Index->new({engine=>$self,offset=>1})->size,
 897     );
 898
 899     my $return = "";
 900
 901     # Header values
 902     $return .= "NumTxns: " . $self->num_txns . $/;
 903
 904     # Read the free sector chains
 905     my %sectors;
 906     foreach my $multiple ( 0 .. 2 ) {
 907         $return .= "Chains($types{$multiple}):";
 908         my $old_loc = $self->chains_loc + $multiple * $self->byte_size;
 909         while ( 1 ) {
 910             my $loc = unpack(
 911                 $StP{$self->byte_size},
 912                 $self->storage->read_at( $old_loc, $self->byte_size ),
 913             );
 914
 915             # We're now out of free sectors of this kind.
 916             unless ( $loc ) {
 917                 last;
 918             }
 919
 920             $sectors{ $types{$multiple} }{ $loc } = undef;
 921             $old_loc = $loc + SIG_SIZE + $STALE_SIZE;
 922             $return .= " $loc";
 923         }
 924         $return .= $/;
 925     }
 926
 927     SECTOR:
 928     while ( $spot < $self->storage->{end} ) {
 929         # Read each sector in order.
 930         my $sector = $self->_load_sector( $spot );
 931         if ( !$sector ) {
 932             # Find it in the free-sectors that were found already
 933             foreach my $type ( keys %sectors ) {
 934                 if ( exists $sectors{$type}{$spot} ) {
 935                     my $size = $sizes{$type};
 936                     $return .= sprintf "%08d: %s %04d\n", $spot, 'F' . $type, $size;
 937                     $spot += $size;
 938                     next SECTOR;
 939                 }
 940             }
 941
 942             die "********\n$return\nDidn't find free sector for $spot in chains\n********\n";
 943         }
 944         else {
 945             $return .= sprintf "%08d: %s  %04d", $spot, $sector->type, $sector->size;
 946             if ( $sector->type eq 'D' ) {
 947                 $return .= ' ' . $sector->data;
 948             }
 949             elsif ( $sector->type eq 'A' || $sector->type eq 'H' ) {
 950                 $return .= ' REF: ' . $sector->get_refcount;
 951             }
 952             elsif ( $sector->type eq 'B' ) {
 953                 foreach my $bucket ( $sector->chopped_up ) {
 954                     $return .= "\n    ";
 955                     $return .= sprintf "%08d", unpack($StP{$self->byte_size},
 956                         substr( $bucket->[-1], $self->hash_size, $self->byte_size),
 957                     );
 958                     my $l = unpack( $StP{$self->byte_size},
 959                         substr( $bucket->[-1],
 960                             $self->hash_size + $self->byte_size,
 961                             $self->byte_size,
 962                         ),
 963                     );
 964                     $return .= sprintf " %08d", $l;
 965                     foreach my $txn ( 0 .. $self->num_txns - 2 ) {
 966                         my $l = unpack( $StP{$self->byte_size},
 967                             substr( $bucket->[-1],
 968                                 $self->hash_size + 2 * $self->byte_size + $txn * ($self->byte_size + $STALE_SIZE),
 969                                 $self->byte_size,
 970                             ),
 971                         );
 972                         $return .= sprintf " %08d", $l;
 973                     }
 974                 }
 975             }
 976             $return .= $/;
 977
 978             $spot += $sector->size;
 979         }
 980     }
 981
 982     return $return;
 983 }
 984
 985 ################################################################################
 986
 987 package DBM::Deep::Iterator;
 988
 989 sub new {
 990     my $class = shift;
 991     my ($args) = @_;
 992
 993     my $self = bless {
 994         breadcrumbs => [],
 995         engine      => $args->{engine},
 996         base_offset => $args->{base_offset},
 997     }, $class;
 998
 999     Scalar::Util::weaken( $self->{engine} );
1000
1001     return $self;
1002 }
1003
1004 sub reset { $_[0]{breadcrumbs} = [] }
1005
1006 sub get_sector_iterator {
1007     my $self = shift;
1008     my ($loc) = @_;
1009
1010     my $sector = $self->{engine}->_load_sector( $loc )
1011         or return;
1012
1013     if ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
1014         return DBM::Deep::Iterator::Index->new({
1015             iterator => $self,
1016             sector   => $sector,
1017         });
1018     }
1019     elsif ( $sector->isa( 'DBM::Deep::Engine::Sector::BucketList' ) ) {
1020         return DBM::Deep::Iterator::BucketList->new({
1021             iterator => $self,
1022             sector   => $sector,
1023         });
1024     }
1025
1026     DBM::Deep->_throw_error( "get_sector_iterator(): Why did $loc make a $sector?" );
1027 }
1028
1029 sub get_next_key {
1030     my $self = shift;
1031     my ($obj) = @_;
1032
1033     my $crumbs = $self->{breadcrumbs};
1034     my $e = $self->{engine};
1035
1036     unless ( @$crumbs ) {
1037         # This will be a Reference sector
1038         my $sector = $e->_load_sector( $self->{base_offset} )
1039             # If no sector is found, thist must have been deleted from under us.
1040             or return;
1041
1042         if ( $sector->staleness != $obj->_staleness ) {
1043             return;
1044         }
1045
1046         my $loc = $sector->get_blist_loc
1047             or return;
1048
1049         push @$crumbs, $self->get_sector_iterator( $loc );
1050     }
1051
1052     FIND_NEXT_KEY: {
1053         # We're at the end.
1054         unless ( @$crumbs ) {
1055             $self->reset;
1056             return;
1057         }
1058
1059         my $iterator = $crumbs->[-1];
1060
1061         # This level is done.
1062         if ( $iterator->at_end ) {
1063             pop @$crumbs;
1064             redo FIND_NEXT_KEY;
1065         }
1066
1067         if ( $iterator->isa( 'DBM::Deep::Iterator::Index' ) ) {
1068             # If we don't have any more, it will be caught at the
1069             # prior check.
1070             if ( my $next = $iterator->get_next_iterator ) {
1071                 push @$crumbs, $next;
1072             }
1073             redo FIND_NEXT_KEY;
1074         }
1075
1076         unless ( $iterator->isa( 'DBM::Deep::Iterator::BucketList' ) ) {
1077             DBM::Deep->_throw_error(
1078                 "Should have a bucketlist iterator here - instead have $iterator"
1079             );
1080         }
1081
1082         # At this point, we have a BucketList iterator
1083         my $key = $iterator->get_next_key;
1084         if ( defined $key ) {
1085             return $key;
1086         }
1087         #XXX else { $iterator->set_to_end() } ?
1088
1089         # We hit the end of the bucketlist iterator, so redo
1090         redo FIND_NEXT_KEY;
1091     }
1092
1093     DBM::Deep->_throw_error( "get_next_key(): How did we get here?" );
1094 }
1095
1096 package DBM::Deep::Iterator::Index;
1097
1098 sub new {
1099     my $self = bless $_[1] => $_[0];
1100     $self->{curr_index} = 0;
1101     return $self;
1102 }
1103
1104 sub at_end {
1105     my $self = shift;
1106     return $self->{curr_index} >= $self->{iterator}{engine}->hash_chars;
1107 }
1108
1109 sub get_next_iterator {
1110     my $self = shift;
1111
1112     my $loc;
1113     while ( !$loc ) {
1114         return if $self->at_end;
1115         $loc = $self->{sector}->get_entry( $self->{curr_index}++ );
1116     }
1117
1118     return $self->{iterator}->get_sector_iterator( $loc );
1119 }
1120
1121 package DBM::Deep::Iterator::BucketList;
1122
1123 sub new {
1124     my $self = bless $_[1] => $_[0];
1125     $self->{curr_index} = 0;
1126     return $self;
1127 }
1128
1129 sub at_end {
1130     my $self = shift;
1131     return $self->{curr_index} >= $self->{iterator}{engine}->max_buckets;
1132 }
1133
1134 sub get_next_key {
1135     my $self = shift;
1136
1137     return if $self->at_end;
1138
1139     my $idx = $self->{curr_index}++;
1140
1141     my $data_loc = $self->{sector}->get_data_location_for({
1142         allow_head => 1,
1143         idx        => $idx,
1144     }) or return;
1145
1146     #XXX Do we want to add corruption checks here?
1147     return $self->{sector}->get_key_for( $idx )->data;
1148 }
1149
1150 package DBM::Deep::Engine::Sector;
1151
1152 sub new {
1153     my $self = bless $_[1], $_[0];
1154     Scalar::Util::weaken( $self->{engine} );
1155     $self->_init;
1156     return $self;
1157 }
1158
1159 #sub _init {}
1160 #sub clone { DBM::Deep->_throw_error( "Must be implemented in the child class" ); }
1161
1162 sub engine { $_[0]{engine} }
1163 sub offset { $_[0]{offset} }
1164 sub type   { $_[0]{type} }
1165
1166 sub base_size {
1167    my $self = shift;
1168    return $self->engine->SIG_SIZE + $STALE_SIZE;
1169 }
1170
1171 sub free {
1172     my $self = shift;
1173
1174     my $e = $self->engine;
1175
1176     $e->storage->print_at( $self->offset, $e->SIG_FREE );
1177     # Skip staleness counter
1178     $e->storage->print_at( $self->offset + $self->base_size,
1179         chr(0) x ($self->size - $self->base_size),
1180     );
1181
1182     my $free_meth = $self->free_meth;
1183     $e->$free_meth( $self->offset, $self->size );
1184
1185     return;
1186 }
1187
1188 package DBM::Deep::Engine::Sector::Data;
1189
1190 our @ISA = qw( DBM::Deep::Engine::Sector );
1191
1192 # This is in bytes
1193 sub size { $_[0]{engine}->data_sector_size }
1194 sub free_meth { return '_add_free_data_sector' }
1195
1196 sub clone {
1197     my $self = shift;
1198     return ref($self)->new({
1199         engine => $self->engine,
1200         type   => $self->type,
1201         data   => $self->data,
1202     });
1203 }
1204
1205 package DBM::Deep::Engine::Sector::Scalar;
1206
1207 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1208
1209 sub free {
1210     my $self = shift;
1211
1212     my $chain_loc = $self->chain_loc;
1213
1214     $self->SUPER::free();
1215
1216     if ( $chain_loc ) {
1217         $self->engine->_load_sector( $chain_loc )->free;
1218     }
1219
1220     return;
1221 }
1222
1223 sub type { $_[0]{engine}->SIG_DATA }
1224 sub _init {
1225     my $self = shift;
1226
1227     my $engine = $self->engine;
1228
1229     unless ( $self->offset ) {
1230         my $data_section = $self->size - $self->base_size - $engine->byte_size - 1;
1231
1232         $self->{offset} = $engine->_request_data_sector( $self->size );
1233
1234         my $data = delete $self->{data};
1235         my $dlen = length $data;
1236         my $continue = 1;
1237         my $curr_offset = $self->offset;
1238         while ( $continue ) {
1239
1240             my $next_offset = 0;
1241
1242             my ($leftover, $this_len, $chunk);
1243             if ( $dlen > $data_section ) {
1244                 $leftover = 0;
1245                 $this_len = $data_section;
1246                 $chunk = substr( $data, 0, $this_len );
1247
1248                 $dlen -= $data_section;
1249                 $next_offset = $engine->_request_data_sector( $self->size );
1250                 $data = substr( $data, $this_len );
1251             }
1252             else {
1253                 $leftover = $data_section - $dlen;
1254                 $this_len = $dlen;
1255                 $chunk = $data;
1256
1257                 $continue = 0;
1258             }
1259
1260             $engine->storage->print_at( $curr_offset, $self->type ); # Sector type
1261             # Skip staleness
1262             $engine->storage->print_at( $curr_offset + $self->base_size,
1263                 pack( $StP{$engine->byte_size}, $next_offset ),  # Chain loc
1264                 pack( $StP{1}, $this_len ),                      # Data length
1265                 $chunk,                                          # Data to be stored in this sector
1266                 chr(0) x $leftover,                              # Zero-fill the rest
1267             );
1268
1269             $curr_offset = $next_offset;
1270         }
1271
1272         return;
1273     }
1274 }
1275
1276 sub data_length {
1277     my $self = shift;
1278
1279     my $buffer = $self->engine->storage->read_at(
1280         $self->offset + $self->base_size + $self->engine->byte_size, 1
1281     );
1282
1283     return unpack( $StP{1}, $buffer );
1284 }
1285
1286 sub chain_loc {
1287     my $self = shift;
1288     return unpack(
1289         $StP{$self->engine->byte_size},
1290         $self->engine->storage->read_at(
1291             $self->offset + $self->base_size,
1292             $self->engine->byte_size,
1293         ),
1294     );
1295 }
1296
1297 sub data {
1298     my $self = shift;
1299
1300     my $data;
1301     while ( 1 ) {
1302         my $chain_loc = $self->chain_loc;
1303
1304         $data .= $self->engine->storage->read_at(
1305             $self->offset + $self->base_size + $self->engine->byte_size + 1, $self->data_length,
1306         );
1307
1308         last unless $chain_loc;
1309
1310         $self = $self->engine->_load_sector( $chain_loc );
1311     }
1312
1313     return $data;
1314 }
1315
1316 package DBM::Deep::Engine::Sector::Null;
1317
1318 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1319
1320 sub type { $_[0]{engine}->SIG_NULL }
1321 sub data_length { 0 }
1322 sub data { return }
1323
1324 sub _init {
1325     my $self = shift;
1326
1327     my $engine = $self->engine;
1328
1329     unless ( $self->offset ) {
1330         my $leftover = $self->size - $self->base_size - 1 * $engine->byte_size - 1;
1331
1332         $self->{offset} = $engine->_request_data_sector( $self->size );
1333         $engine->storage->print_at( $self->offset, $self->type ); # Sector type
1334         # Skip staleness counter
1335         $engine->storage->print_at( $self->offset + $self->base_size,
1336             pack( $StP{$engine->byte_size}, 0 ),  # Chain loc
1337             pack( $StP{1}, $self->data_length ),  # Data length
1338             chr(0) x $leftover,                   # Zero-fill the rest
1339         );
1340
1341         return;
1342     }
1343 }
1344
1345 package DBM::Deep::Engine::Sector::Reference;
1346
1347 # XXX This is where we're going to add the refcounting
1348
1349 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1350
1351 sub _init {
1352     my $self = shift;
1353
1354     my $e = $self->engine;
1355
1356     unless ( $self->offset ) {
1357         my $classname = Scalar::Util::blessed( delete $self->{data} );
1358         my $leftover = $self->size - $self->base_size - 3 * $e->byte_size;
1359
1360         my $class_offset = 0;
1361         if ( defined $classname ) {
1362             my $class_sector = DBM::Deep::Engine::Sector::Scalar->new({
1363                 engine => $e,
1364                 data   => $classname,
1365             });
1366             $class_offset = $class_sector->offset;
1367         }
1368
1369         $self->{offset} = $e->_request_data_sector( $self->size );
1370         $e->storage->print_at( $self->offset, $self->type ); # Sector type
1371         # Skip staleness counter
1372         $e->storage->print_at( $self->offset + $self->base_size,
1373             pack( $StP{$e->byte_size}, 0 ),             # Index/BList loc
1374             pack( $StP{$e->byte_size}, $class_offset ), # Classname loc
1375             pack( $StP{$e->byte_size}, 1 ),             # Initial refcount
1376             chr(0) x $leftover,                         # Zero-fill the rest
1377         );
1378     }
1379     else {
1380         $self->{type} = $e->storage->read_at( $self->offset, 1 );
1381     }
1382
1383     $self->{staleness} = unpack(
1384         $StP{$STALE_SIZE},
1385         $e->storage->read_at( $self->offset + $e->SIG_SIZE, $STALE_SIZE ),
1386     );
1387
1388     return;
1389 }
1390
1391 sub staleness { $_[0]{staleness} }
1392
1393 sub get_data_for {
1394     my $self = shift;
1395     my ($args) = @_;
1396
1397     # Assume that the head is not allowed unless otherwise specified.
1398     $args->{allow_head} = 0 unless exists $args->{allow_head};
1399
1400     # Assume we don't create a new blist location unless otherwise specified.
1401     $args->{create} = 0 unless exists $args->{create};
1402
1403     my $blist = $self->get_bucket_list({
1404         key_md5 => $args->{key_md5},
1405         key => $args->{key},
1406         create  => $args->{create},
1407     });
1408     return unless $blist && $blist->{found};
1409
1410     # At this point, $blist knows where the md5 is. What it -doesn't- know yet
1411     # is whether or not this transaction has this key. That's part of the next
1412     # function call.
1413     my $location = $blist->get_data_location_for({
1414         allow_head => $args->{allow_head},
1415     }) or return;
1416
1417     return $self->engine->_load_sector( $location );
1418 }
1419
1420 sub write_data {
1421     my $self = shift;
1422     my ($args) = @_;
1423
1424     my $blist = $self->get_bucket_list({
1425         key_md5 => $args->{key_md5},
1426         key => $args->{key},
1427         create  => 1,
1428     }) or DBM::Deep->_throw_error( "How did write_data fail (no blist)?!" );
1429
1430     # Handle any transactional bookkeeping.
1431     if ( $self->engine->trans_id ) {
1432         if ( ! $blist->has_md5 ) {
1433             $blist->mark_deleted({
1434                 trans_id => 0,
1435             });
1436         }
1437     }
1438     else {
1439         my @trans_ids = $self->engine->get_running_txn_ids;
1440         if ( $blist->has_md5 ) {
1441             if ( @trans_ids ) {
1442                 my $old_value = $blist->get_data_for;
1443                 foreach my $other_trans_id ( @trans_ids ) {
1444                     next if $blist->get_data_location_for({
1445                         trans_id   => $other_trans_id,
1446                         allow_head => 0,
1447                     });
1448                     $blist->write_md5({
1449                         trans_id => $other_trans_id,
1450                         key      => $args->{key},
1451                         key_md5  => $args->{key_md5},
1452                         value    => $old_value->clone,
1453                     });
1454                 }
1455             }
1456         }
1457         else {
1458             if ( @trans_ids ) {
1459                 foreach my $other_trans_id ( @trans_ids ) {
1460                     #XXX This doesn't seem to possible to ever happen . . .
1461                     next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1462                     $blist->mark_deleted({
1463                         trans_id => $other_trans_id,
1464                     });
1465                 }
1466             }
1467         }
1468     }
1469
1470     #XXX Is this safe to do transactionally?
1471     # Free the place we're about to write to.
1472     if ( $blist->get_data_location_for({ allow_head => 0 }) ) {
1473         $blist->get_data_for({ allow_head => 0 })->free;
1474     }
1475
1476     $blist->write_md5({
1477         key      => $args->{key},
1478         key_md5  => $args->{key_md5},
1479         value    => $args->{value},
1480     });
1481 }
1482
1483 sub delete_key {
1484     my $self = shift;
1485     my ($args) = @_;
1486
1487     # XXX What should happen if this fails?
1488     my $blist = $self->get_bucket_list({
1489         key_md5 => $args->{key_md5},
1490     }) or DBM::Deep->_throw_error( "How did delete_key fail (no blist)?!" );
1491
1492     # Save the location so that we can free the data
1493     my $location = $blist->get_data_location_for({
1494         allow_head => 0,
1495     });
1496     my $old_value = $location && $self->engine->_load_sector( $location );
1497
1498     my @trans_ids = $self->engine->get_running_txn_ids;
1499
1500     # If we're the HEAD and there are running txns, then we need to clone this value to the other
1501     # transactions to preserve Isolation.
1502     if ( $self->engine->trans_id == 0 ) {
1503         if ( @trans_ids ) {
1504             foreach my $other_trans_id ( @trans_ids ) {
1505                 next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1506                 $blist->write_md5({
1507                     trans_id => $other_trans_id,
1508                     key      => $args->{key},
1509                     key_md5  => $args->{key_md5},
1510                     value    => $old_value->clone,
1511                 });
1512             }
1513         }
1514     }
1515
1516     my $data;
1517     if ( @trans_ids ) {
1518         $blist->mark_deleted( $args );
1519
1520         if ( $old_value ) {
1521             $data = $old_value->data;
1522             $old_value->free;
1523         }
1524     }
1525     else {
1526         $data = $blist->delete_md5( $args );
1527     }
1528
1529     return $data;
1530 }
1531
1532 sub get_blist_loc {
1533     my $self = shift;
1534
1535     my $e = $self->engine;
1536     my $blist_loc = $e->storage->read_at( $self->offset + $self->base_size, $e->byte_size );
1537     return unpack( $StP{$e->byte_size}, $blist_loc );
1538 }
1539
1540 sub get_bucket_list {
1541     my $self = shift;
1542     my ($args) = @_;
1543     $args ||= {};
1544
1545     # XXX Add in check here for recycling?
1546
1547     my $engine = $self->engine;
1548
1549     my $blist_loc = $self->get_blist_loc;
1550
1551     # There's no index or blist yet
1552     unless ( $blist_loc ) {
1553         return unless $args->{create};
1554
1555         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1556             engine  => $engine,
1557             key_md5 => $args->{key_md5},
1558         });
1559
1560         $engine->storage->print_at( $self->offset + $self->base_size,
1561             pack( $StP{$engine->byte_size}, $blist->offset ),
1562         );
1563
1564         return $blist;
1565     }
1566
1567     my $sector = $engine->_load_sector( $blist_loc )
1568         or DBM::Deep->_throw_error( "Cannot read sector at $blist_loc in get_bucket_list()" );
1569     my $i = 0;
1570     my $last_sector = undef;
1571     while ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
1572         $blist_loc = $sector->get_entry( ord( substr( $args->{key_md5}, $i++, 1 ) ) );
1573         $last_sector = $sector;
1574         if ( $blist_loc ) {
1575             $sector = $engine->_load_sector( $blist_loc )
1576                 or DBM::Deep->_throw_error( "Cannot read sector at $blist_loc in get_bucket_list()" );
1577         }
1578         else {
1579             $sector = undef;
1580             last;
1581         }
1582     }
1583
1584     # This means we went through the Index sector(s) and found an empty slot
1585     unless ( $sector ) {
1586         return unless $args->{create};
1587
1588         DBM::Deep->_throw_error( "No last_sector when attempting to build a new entry" )
1589             unless $last_sector;
1590
1591         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1592             engine  => $engine,
1593             key_md5 => $args->{key_md5},
1594         });
1595
1596         $last_sector->set_entry( ord( substr( $args->{key_md5}, $i - 1, 1 ) ) => $blist->offset );
1597
1598         return $blist;
1599     }
1600
1601     $sector->find_md5( $args->{key_md5} );
1602
1603     # See whether or not we need to reindex the bucketlist
1604     # Yes, the double-braces are there for a reason. if() doesn't create a redo-able block,
1605     # so we have to create a bare block within the if() for redo-purposes. Patch and idea
1606     # submitted by sprout@cpan.org. -RobK, 2008-01-09
1607     if ( !$sector->has_md5 && $args->{create} && $sector->{idx} == -1 ) {{
1608         my $redo;
1609
1610         my $new_index = DBM::Deep::Engine::Sector::Index->new({
1611             engine => $engine,
1612         });
1613
1614         my %blist_cache;
1615         #XXX q.v. the comments for this function.
1616         foreach my $entry ( $sector->chopped_up ) {
1617             my ($spot, $md5) = @{$entry};
1618             my $idx = ord( substr( $md5, $i, 1 ) );
1619
1620             # XXX This is inefficient
1621             my $blist = $blist_cache{$idx}
1622                 ||= DBM::Deep::Engine::Sector::BucketList->new({
1623                     engine => $engine,
1624                 });
1625
1626             $new_index->set_entry( $idx => $blist->offset );
1627
1628             my $new_spot = $blist->write_at_next_open( $md5 );
1629             $engine->reindex_entry( $spot => $new_spot );
1630         }
1631
1632         # Handle the new item separately.
1633         {
1634             my $idx = ord( substr( $args->{key_md5}, $i, 1 ) );
1635
1636             # If all the previous blist's items have been thrown into one
1637             # blist and the new item belongs in there too, we need
1638             # another index.
1639             if ( keys %blist_cache == 1 and each %blist_cache == $idx ) {
1640                 ++$i, ++$redo;
1641             } else {
1642                 my $blist = $blist_cache{$idx}
1643                     ||= DBM::Deep::Engine::Sector::BucketList->new({
1644                         engine => $engine,
1645                     });
1646
1647                 $new_index->set_entry( $idx => $blist->offset );
1648
1649                 #XXX THIS IS HACKY!
1650                 $blist->find_md5( $args->{key_md5} );
1651                 $blist->write_md5({
1652                     key     => $args->{key},
1653                     key_md5 => $args->{key_md5},
1654                     value   => DBM::Deep::Engine::Sector::Null->new({
1655                         engine => $engine,
1656                         data   => undef,
1657                     }),
1658                 });
1659             }
1660 #            my $blist = $blist_cache{$idx}
1661 #                ||= DBM::Deep::Engine::Sector::BucketList->new({
1662 #                    engine => $engine,
1663 #                });
1664 #
1665 #            $new_index->set_entry( $idx => $blist->offset );
1666 #
1667 #            #XXX THIS IS HACKY!
1668 #            $blist->find_md5( $args->{key_md5} );
1669 #            $blist->write_md5({
1670 #                key     => $args->{key},
1671 #                key_md5 => $args->{key_md5},
1672 #                value   => DBM::Deep::Engine::Sector::Null->new({
1673 #                    engine => $engine,
1674 #                    data   => undef,
1675 #                }),
1676 #            });
1677         }
1678
1679         if ( $last_sector ) {
1680             $last_sector->set_entry(
1681                 ord( substr( $args->{key_md5}, $i - 1, 1 ) ),
1682                 $new_index->offset,
1683             );
1684         } else {
1685             $engine->storage->print_at( $self->offset + $self->base_size,
1686                 pack( $StP{$engine->byte_size}, $new_index->offset ),
1687             );
1688         }
1689
1690         $sector->clear;
1691         $sector->free;
1692
1693         if ( $redo ) {
1694             (undef, $sector) = %blist_cache;
1695             $last_sector = $new_index;
1696             redo;
1697         }
1698
1699         $sector = $blist_cache{ ord( substr( $args->{key_md5}, $i, 1 ) ) };
1700         $sector->find_md5( $args->{key_md5} );
1701     }}
1702
1703     return $sector;
1704 }
1705
1706 sub get_class_offset {
1707     my $self = shift;
1708
1709     my $e = $self->engine;
1710     return unpack(
1711         $StP{$e->byte_size},
1712         $e->storage->read_at(
1713             $self->offset + $self->base_size + 1 * $e->byte_size, $e->byte_size,
1714         ),
1715     );
1716 }
1717
1718 sub get_classname {
1719     my $self = shift;
1720
1721     my $class_offset = $self->get_class_offset;
1722
1723     return unless $class_offset;
1724
1725     return $self->engine->_load_sector( $class_offset )->data;
1726 }
1727
1728 sub data {
1729     my $self = shift;
1730
1731     unless ( $self->engine->cache->{ $self->offset } ) {
1732         my $new_obj = DBM::Deep->new({
1733             type        => $self->type,
1734             base_offset => $self->offset,
1735             staleness   => $self->staleness,
1736             storage     => $self->engine->storage,
1737             engine      => $self->engine,
1738         });
1739
1740         if ( $self->engine->storage->{autobless} ) {
1741             my $classname = $self->get_classname;
1742             if ( defined $classname ) {
1743                 bless $new_obj, $classname;
1744             }
1745         }
1746
1747         $self->engine->cache->{$self->offset} = $new_obj;
1748     }
1749     return $self->engine->cache->{$self->offset};
1750 }
1751
1752 sub free {
1753     my $self = shift;
1754
1755     # We're not ready to be removed yet.
1756     if ( $self->decrement_refcount > 0 ) {
1757         return;
1758     }
1759
1760     # Rebless the object into DBM::Deep::Null.
1761     eval { %{ $self->engine->cache->{ $self->offset } } = (); };
1762     eval { @{ $self->engine->cache->{ $self->offset } } = (); };
1763     bless $self->engine->cache->{ $self->offset }, 'DBM::Deep::Null';
1764     delete $self->engine->cache->{ $self->offset };
1765
1766     my $blist_loc = $self->get_blist_loc;
1767     $self->engine->_load_sector( $blist_loc )->free if $blist_loc;
1768
1769     my $class_loc = $self->get_class_offset;
1770     $self->engine->_load_sector( $class_loc )->free if $class_loc;
1771
1772     $self->SUPER::free();
1773 }
1774
1775 sub increment_refcount {
1776     my $self = shift;
1777
1778     my $refcount = $self->get_refcount;
1779
1780     $refcount++;
1781
1782     $self->write_refcount( $refcount );
1783
1784     return $refcount;
1785 }
1786
1787 sub decrement_refcount {
1788     my $self = shift;
1789
1790     my $refcount = $self->get_refcount;
1791
1792     $refcount--;
1793
1794     $self->write_refcount( $refcount );
1795
1796     return $refcount;
1797 }
1798
1799 sub get_refcount {
1800     my $self = shift;
1801
1802     my $e = $self->engine;
1803     return unpack(
1804         $StP{$e->byte_size},
1805         $e->storage->read_at(
1806             $self->offset + $self->base_size + 2 * $e->byte_size, $e->byte_size,
1807         ),
1808     );
1809 }
1810
1811 sub write_refcount {
1812     my $self = shift;
1813     my ($num) = @_;
1814
1815     my $e = $self->engine;
1816     $e->storage->print_at(
1817         $self->offset + $self->base_size + 2 * $e->byte_size,
1818         pack( $StP{$e->byte_size}, $num ),
1819     );
1820 }
1821
1822 package DBM::Deep::Engine::Sector::BucketList;
1823
1824 our @ISA = qw( DBM::Deep::Engine::Sector );
1825
1826 sub _init {
1827     my $self = shift;
1828
1829     my $engine = $self->engine;
1830
1831     unless ( $self->offset ) {
1832         my $leftover = $self->size - $self->base_size;
1833
1834         $self->{offset} = $engine->_request_blist_sector( $self->size );
1835         $engine->storage->print_at( $self->offset, $engine->SIG_BLIST ); # Sector type
1836         # Skip staleness counter
1837         $engine->storage->print_at( $self->offset + $self->base_size,
1838             chr(0) x $leftover, # Zero-fill the data
1839         );
1840     }
1841
1842     if ( $self->{key_md5} ) {
1843         $self->find_md5;
1844     }
1845
1846     return $self;
1847 }
1848
1849 sub clear {
1850     my $self = shift;
1851     $self->engine->storage->print_at( $self->offset + $self->base_size,
1852         chr(0) x ($self->size - $self->base_size), # Zero-fill the data
1853     );
1854 }
1855
1856 sub size {
1857     my $self = shift;
1858     unless ( $self->{size} ) {
1859         my $e = $self->engine;
1860         # Base + numbuckets * bucketsize
1861         $self->{size} = $self->base_size + $e->max_buckets * $self->bucket_size;
1862     }
1863     return $self->{size};
1864 }
1865
1866 sub free_meth { return '_add_free_blist_sector' }
1867
1868 sub free {
1869     my $self = shift;
1870
1871     my $e = $self->engine;
1872     foreach my $bucket ( $self->chopped_up ) {
1873         my $rest = $bucket->[-1];
1874
1875         # Delete the keysector
1876         my $l = unpack( $StP{$e->byte_size}, substr( $rest, $e->hash_size, $e->byte_size ) );
1877         my $s = $e->_load_sector( $l ); $s->free if $s;
1878
1879         # Delete the HEAD sector
1880         $l = unpack( $StP{$e->byte_size},
1881             substr( $rest,
1882                 $e->hash_size + $e->byte_size,
1883                 $e->byte_size,
1884             ),
1885         );
1886         $s = $e->_load_sector( $l ); $s->free if $s;
1887
1888         foreach my $txn ( 0 .. $e->num_txns - 2 ) {
1889             my $l = unpack( $StP{$e->byte_size},
1890                 substr( $rest,
1891                     $e->hash_size + 2 * $e->byte_size + $txn * ($e->byte_size + $STALE_SIZE),
1892                     $e->byte_size,
1893                 ),
1894             );
1895             my $s = $e->_load_sector( $l ); $s->free if $s;
1896         }
1897     }
1898
1899     $self->SUPER::free();
1900 }
1901
1902 sub bucket_size {
1903     my $self = shift;
1904     unless ( $self->{bucket_size} ) {
1905         my $e = $self->engine;
1906         # Key + head (location) + transactions (location + staleness-counter)
1907         my $location_size = $e->byte_size + $e->byte_size + ($e->num_txns - 1) * ($e->byte_size + $STALE_SIZE);
1908         $self->{bucket_size} = $e->hash_size + $location_size;
1909     }
1910     return $self->{bucket_size};
1911 }
1912
1913 # XXX This is such a poor hack. I need to rethink this code.
1914 sub chopped_up {
1915     my $self = shift;
1916
1917     my $e = $self->engine;
1918
1919     my @buckets;
1920     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1921         my $spot = $self->offset + $self->base_size + $idx * $self->bucket_size;
1922         my $md5 = $e->storage->read_at( $spot, $e->hash_size );
1923
1924         #XXX If we're chopping, why would we ever have the blank_md5?
1925         last if $md5 eq $e->blank_md5;
1926
1927         my $rest = $e->storage->read_at( undef, $self->bucket_size - $e->hash_size );
1928         push @buckets, [ $spot, $md5 . $rest ];
1929     }
1930
1931     return @buckets;
1932 }
1933
1934 sub write_at_next_open {
1935     my $self = shift;
1936     my ($entry) = @_;
1937
1938     #XXX This is such a hack!
1939     $self->{_next_open} = 0 unless exists $self->{_next_open};
1940
1941     my $spot = $self->offset + $self->base_size + $self->{_next_open}++ * $self->bucket_size;
1942     $self->engine->storage->print_at( $spot, $entry );
1943
1944     return $spot;
1945 }
1946
1947 sub has_md5 {
1948     my $self = shift;
1949     unless ( exists $self->{found} ) {
1950         $self->find_md5;
1951     }
1952     return $self->{found};
1953 }
1954
1955 sub find_md5 {
1956     my $self = shift;
1957
1958     $self->{found} = undef;
1959     $self->{idx}   = -1;
1960
1961     if ( @_ ) {
1962         $self->{key_md5} = shift;
1963     }
1964
1965     # If we don't have an MD5, then what are we supposed to do?
1966     unless ( exists $self->{key_md5} ) {
1967         DBM::Deep->_throw_error( "Cannot find_md5 without a key_md5 set" );
1968     }
1969
1970     my $e = $self->engine;
1971     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1972         my $potential = $e->storage->read_at(
1973             $self->offset + $self->base_size + $idx * $self->bucket_size, $e->hash_size,
1974         );
1975
1976         if ( $potential eq $e->blank_md5 ) {
1977             $self->{idx} = $idx;
1978             return;
1979         }
1980
1981         if ( $potential eq $self->{key_md5} ) {
1982             $self->{found} = 1;
1983             $self->{idx} = $idx;
1984             return;
1985         }
1986     }
1987
1988     return;
1989 }
1990
1991 sub write_md5 {
1992     my $self = shift;
1993     my ($args) = @_;
1994
1995     DBM::Deep->_throw_error( "write_md5: no key" ) unless exists $args->{key};
1996     DBM::Deep->_throw_error( "write_md5: no key_md5" ) unless exists $args->{key_md5};
1997     DBM::Deep->_throw_error( "write_md5: no value" ) unless exists $args->{value};
1998
1999     my $engine = $self->engine;
2000
2001     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
2002
2003     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2004     $engine->add_entry( $args->{trans_id}, $spot );
2005
2006     unless ($self->{found}) {
2007         my $key_sector = DBM::Deep::Engine::Sector::Scalar->new({
2008             engine => $engine,
2009             data   => $args->{key},
2010         });
2011
2012         $engine->storage->print_at( $spot,
2013             $args->{key_md5},
2014             pack( $StP{$engine->byte_size}, $key_sector->offset ),
2015         );
2016     }
2017
2018     my $loc = $spot
2019       + $engine->hash_size
2020       + $engine->byte_size;
2021
2022     if ( $args->{trans_id} ) {
2023         $loc += $engine->byte_size + ($args->{trans_id} - 1) * ( $engine->byte_size + $STALE_SIZE );
2024
2025         $engine->storage->print_at( $loc,
2026             pack( $StP{$engine->byte_size}, $args->{value}->offset ),
2027             pack( $StP{$STALE_SIZE}, $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
2028         );
2029     }
2030     else {
2031         $engine->storage->print_at( $loc,
2032             pack( $StP{$engine->byte_size}, $args->{value}->offset ),
2033         );
2034     }
2035 }
2036
2037 sub mark_deleted {
2038     my $self = shift;
2039     my ($args) = @_;
2040     $args ||= {};
2041
2042     my $engine = $self->engine;
2043
2044     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
2045
2046     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2047     $engine->add_entry( $args->{trans_id}, $spot );
2048
2049     my $loc = $spot
2050       + $engine->hash_size
2051       + $engine->byte_size;
2052
2053     if ( $args->{trans_id} ) {
2054         $loc += $engine->byte_size + ($args->{trans_id} - 1) * ( $engine->byte_size + $STALE_SIZE );
2055
2056         $engine->storage->print_at( $loc,
2057             pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
2058             pack( $StP{$STALE_SIZE}, $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
2059         );
2060     }
2061     else {
2062         $engine->storage->print_at( $loc,
2063             pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
2064         );
2065     }
2066
2067 }
2068
2069 sub delete_md5 {
2070     my $self = shift;
2071     my ($args) = @_;
2072
2073     my $engine = $self->engine;
2074     return undef unless $self->{found};
2075
2076     # Save the location so that we can free the data
2077     my $location = $self->get_data_location_for({
2078         allow_head => 0,
2079     });
2080     my $key_sector = $self->get_key_for;
2081
2082     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2083     $engine->storage->print_at( $spot,
2084         $engine->storage->read_at(
2085             $spot + $self->bucket_size,
2086             $self->bucket_size * ( $engine->max_buckets - $self->{idx} - 1 ),
2087         ),
2088         chr(0) x $self->bucket_size,
2089     );
2090
2091     $key_sector->free;
2092
2093     # XXX Here is where we add the recount decrement check
2094     my $data_sector = $self->engine->_load_sector( $location );
2095     my $data = $data_sector->data;
2096     $data_sector->free;
2097
2098     return $data;
2099 }
2100
2101 sub get_data_location_for {
2102     my $self = shift;
2103     my ($args) = @_;
2104     $args ||= {};
2105
2106     $args->{allow_head} = 0 unless exists $args->{allow_head};
2107     $args->{trans_id}   = $self->engine->trans_id unless exists $args->{trans_id};
2108     $args->{idx}        = $self->{idx} unless exists $args->{idx};
2109
2110     my $e = $self->engine;
2111
2112     my $spot = $self->offset + $self->base_size
2113       + $args->{idx} * $self->bucket_size
2114       + $e->hash_size
2115       + $e->byte_size;
2116
2117     if ( $args->{trans_id} ) {
2118         $spot += $e->byte_size + ($args->{trans_id} - 1) * ( $e->byte_size + $STALE_SIZE );
2119     }
2120
2121     my $buffer = $e->storage->read_at(
2122         $spot,
2123         $e->byte_size + $STALE_SIZE,
2124     );
2125     my ($loc, $staleness) = unpack( $StP{$e->byte_size} . ' ' . $StP{$STALE_SIZE}, $buffer );
2126
2127     # XXX Merge the two if-clauses below
2128     if ( $args->{trans_id} ) {
2129         # We have found an entry that is old, so get rid of it
2130         if ( $staleness != (my $s = $e->get_txn_staleness_counter( $args->{trans_id} ) ) ) {
2131             $e->storage->print_at(
2132                 $spot,
2133                 pack( $StP{$e->byte_size} . ' ' . $StP{$STALE_SIZE}, (0) x 2 ),
2134             );
2135             $loc = 0;
2136         }
2137     }
2138
2139     # If we're in a transaction and we never wrote to this location, try the
2140     # HEAD instead.
2141     if ( $args->{trans_id} && !$loc && $args->{allow_head} ) {
2142         return $self->get_data_location_for({
2143             trans_id   => 0,
2144             allow_head => 1,
2145             idx        => $args->{idx},
2146         });
2147     }
2148
2149     return $loc <= 1 ? 0 : $loc;
2150 }
2151
2152 sub get_data_for {
2153     my $self = shift;
2154     my ($args) = @_;
2155     $args ||= {};
2156
2157     return unless $self->{found};
2158     my $location = $self->get_data_location_for({
2159         allow_head => $args->{allow_head},
2160     });
2161     return $self->engine->_load_sector( $location );
2162 }
2163
2164 sub get_key_for {
2165     my $self = shift;
2166     my ($idx) = @_;
2167     $idx = $self->{idx} unless defined $idx;
2168
2169     if ( $idx >= $self->engine->max_buckets ) {
2170         DBM::Deep->_throw_error( "get_key_for(): Attempting to retrieve $idx" );
2171     }
2172
2173     my $location = $self->engine->storage->read_at(
2174         $self->offset + $self->base_size + $idx * $self->bucket_size + $self->engine->hash_size,
2175         $self->engine->byte_size,
2176     );
2177     $location = unpack( $StP{$self->engine->byte_size}, $location );
2178     DBM::Deep->_throw_error( "get_key_for: No location?" ) unless $location;
2179
2180     return $self->engine->_load_sector( $location );
2181 }
2182
2183 package DBM::Deep::Engine::Sector::Index;
2184
2185 our @ISA = qw( DBM::Deep::Engine::Sector );
2186
2187 sub _init {
2188     my $self = shift;
2189
2190     my $engine = $self->engine;
2191
2192     unless ( $self->offset ) {
2193         my $leftover = $self->size - $self->base_size;
2194
2195         $self->{offset} = $engine->_request_index_sector( $self->size );
2196         $engine->storage->print_at( $self->offset, $engine->SIG_INDEX ); # Sector type
2197         # Skip staleness counter
2198         $engine->storage->print_at( $self->offset + $self->base_size,
2199             chr(0) x $leftover, # Zero-fill the rest
2200         );
2201     }
2202
2203     return $self;
2204 }
2205
2206 #XXX Change here
2207 sub size {
2208     my $self = shift;
2209     unless ( $self->{size} ) {
2210         my $e = $self->engine;
2211         $self->{size} = $self->base_size + $e->byte_size * $e->hash_chars;
2212     }
2213     return $self->{size};
2214 }
2215
2216 sub free_meth { return '_add_free_index_sector' }
2217
2218 sub free {
2219     my $self = shift;
2220     my $e = $self->engine;
2221
2222     for my $i ( 0 .. $e->hash_chars - 1 ) {
2223         my $l = $self->get_entry( $i ) or next;
2224         $e->_load_sector( $l )->free;
2225     }
2226
2227     $self->SUPER::free();
2228 }
2229
2230 sub _loc_for {
2231     my $self = shift;
2232     my ($idx) = @_;
2233     return $self->offset + $self->base_size + $idx * $self->engine->byte_size;
2234 }
2235
2236 sub get_entry {
2237     my $self = shift;
2238     my ($idx) = @_;
2239
2240     my $e = $self->engine;
2241
2242     DBM::Deep->_throw_error( "get_entry: Out of range ($idx)" )
2243         if $idx < 0 || $idx >= $e->hash_chars;
2244
2245     return unpack(
2246         $StP{$e->byte_size},
2247         $e->storage->read_at( $self->_loc_for( $idx ), $e->byte_size ),
2248     );
2249 }
2250
2251 sub set_entry {
2252     my $self = shift;
2253     my ($idx, $loc) = @_;
2254
2255     my $e = $self->engine;
2256
2257     DBM::Deep->_throw_error( "set_entry: Out of range ($idx)" )
2258         if $idx < 0 || $idx >= $e->hash_chars;
2259
2260     $self->engine->storage->print_at(
2261         $self->_loc_for( $idx ),
2262         pack( $StP{$e->byte_size}, $loc ),
2263     );
2264 }
2265
2266 # This was copied from MARCEL's Class::Null. However, I couldn't use it because
2267 # I need an undef value, not an implementation of the Null Class pattern.
2268 package DBM::Deep::Null;
2269
2270 use overload
2271     'bool'   => sub { undef },
2272     '""'     => sub { undef },
2273     '0+'     => sub { undef },
2274     fallback => 1,
2275     nomethod => 'AUTOLOAD';
2276
2277 sub AUTOLOAD { return; }
2278
2279 1;
2280 __END__