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.0008);
   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 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1348
1349 sub _init {
1350     my $self = shift;
1351
1352     my $e = $self->engine;
1353
1354     unless ( $self->offset ) {
1355         my $classname = Scalar::Util::blessed( delete $self->{data} );
1356         my $leftover = $self->size - $self->base_size - 3 * $e->byte_size;
1357
1358         my $class_offset = 0;
1359         if ( defined $classname ) {
1360             my $class_sector = DBM::Deep::Engine::Sector::Scalar->new({
1361                 engine => $e,
1362                 data   => $classname,
1363             });
1364             $class_offset = $class_sector->offset;
1365         }
1366
1367         $self->{offset} = $e->_request_data_sector( $self->size );
1368         $e->storage->print_at( $self->offset, $self->type ); # Sector type
1369         # Skip staleness counter
1370         $e->storage->print_at( $self->offset + $self->base_size,
1371             pack( $StP{$e->byte_size}, 0 ),             # Index/BList loc
1372             pack( $StP{$e->byte_size}, $class_offset ), # Classname loc
1373             pack( $StP{$e->byte_size}, 1 ),             # Initial refcount
1374             chr(0) x $leftover,                         # Zero-fill the rest
1375         );
1376     }
1377     else {
1378         $self->{type} = $e->storage->read_at( $self->offset, 1 );
1379     }
1380
1381     $self->{staleness} = unpack(
1382         $StP{$STALE_SIZE},
1383         $e->storage->read_at( $self->offset + $e->SIG_SIZE, $STALE_SIZE ),
1384     );
1385
1386     return;
1387 }
1388
1389 sub staleness { $_[0]{staleness} }
1390
1391 sub get_data_for {
1392     my $self = shift;
1393     my ($args) = @_;
1394
1395     # Assume that the head is not allowed unless otherwise specified.
1396     $args->{allow_head} = 0 unless exists $args->{allow_head};
1397
1398     # Assume we don't create a new blist location unless otherwise specified.
1399     $args->{create} = 0 unless exists $args->{create};
1400
1401     my $blist = $self->get_bucket_list({
1402         key_md5 => $args->{key_md5},
1403         key => $args->{key},
1404         create  => $args->{create},
1405     });
1406     return unless $blist && $blist->{found};
1407
1408     # At this point, $blist knows where the md5 is. What it -doesn't- know yet
1409     # is whether or not this transaction has this key. That's part of the next
1410     # function call.
1411     my $location = $blist->get_data_location_for({
1412         allow_head => $args->{allow_head},
1413     }) or return;
1414
1415     return $self->engine->_load_sector( $location );
1416 }
1417
1418 sub write_data {
1419     my $self = shift;
1420     my ($args) = @_;
1421
1422     my $blist = $self->get_bucket_list({
1423         key_md5 => $args->{key_md5},
1424         key => $args->{key},
1425         create  => 1,
1426     }) or DBM::Deep->_throw_error( "How did write_data fail (no blist)?!" );
1427
1428     # Handle any transactional bookkeeping.
1429     if ( $self->engine->trans_id ) {
1430         if ( ! $blist->has_md5 ) {
1431             $blist->mark_deleted({
1432                 trans_id => 0,
1433             });
1434         }
1435     }
1436     else {
1437         my @trans_ids = $self->engine->get_running_txn_ids;
1438         if ( $blist->has_md5 ) {
1439             if ( @trans_ids ) {
1440                 my $old_value = $blist->get_data_for;
1441                 foreach my $other_trans_id ( @trans_ids ) {
1442                     next if $blist->get_data_location_for({
1443                         trans_id   => $other_trans_id,
1444                         allow_head => 0,
1445                     });
1446                     $blist->write_md5({
1447                         trans_id => $other_trans_id,
1448                         key      => $args->{key},
1449                         key_md5  => $args->{key_md5},
1450                         value    => $old_value->clone,
1451                     });
1452                 }
1453             }
1454         }
1455         else {
1456             if ( @trans_ids ) {
1457                 foreach my $other_trans_id ( @trans_ids ) {
1458                     #XXX This doesn't seem to possible to ever happen . . .
1459                     next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1460                     $blist->mark_deleted({
1461                         trans_id => $other_trans_id,
1462                     });
1463                 }
1464             }
1465         }
1466     }
1467
1468     #XXX Is this safe to do transactionally?
1469     # Free the place we're about to write to.
1470     if ( $blist->get_data_location_for({ allow_head => 0 }) ) {
1471         $blist->get_data_for({ allow_head => 0 })->free;
1472     }
1473
1474     $blist->write_md5({
1475         key      => $args->{key},
1476         key_md5  => $args->{key_md5},
1477         value    => $args->{value},
1478     });
1479 }
1480
1481 sub delete_key {
1482     my $self = shift;
1483     my ($args) = @_;
1484
1485     # XXX What should happen if this fails?
1486     my $blist = $self->get_bucket_list({
1487         key_md5 => $args->{key_md5},
1488     }) or DBM::Deep->_throw_error( "How did delete_key fail (no blist)?!" );
1489
1490     # Save the location so that we can free the data
1491     my $location = $blist->get_data_location_for({
1492         allow_head => 0,
1493     });
1494     my $old_value = $location && $self->engine->_load_sector( $location );
1495
1496     my @trans_ids = $self->engine->get_running_txn_ids;
1497
1498     # If we're the HEAD and there are running txns, then we need to clone this value to the other
1499     # transactions to preserve Isolation.
1500     if ( $self->engine->trans_id == 0 ) {
1501         if ( @trans_ids ) {
1502             foreach my $other_trans_id ( @trans_ids ) {
1503                 next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1504                 $blist->write_md5({
1505                     trans_id => $other_trans_id,
1506                     key      => $args->{key},
1507                     key_md5  => $args->{key_md5},
1508                     value    => $old_value->clone,
1509                 });
1510             }
1511         }
1512     }
1513
1514     my $data;
1515     if ( @trans_ids ) {
1516         $blist->mark_deleted( $args );
1517
1518         if ( $old_value ) {
1519             $data = $old_value->data;
1520             $old_value->free;
1521         }
1522     }
1523     else {
1524         $data = $blist->delete_md5( $args );
1525     }
1526
1527     return $data;
1528 }
1529
1530 sub get_blist_loc {
1531     my $self = shift;
1532
1533     my $e = $self->engine;
1534     my $blist_loc = $e->storage->read_at( $self->offset + $self->base_size, $e->byte_size );
1535     return unpack( $StP{$e->byte_size}, $blist_loc );
1536 }
1537
1538 sub get_bucket_list {
1539     my $self = shift;
1540     my ($args) = @_;
1541     $args ||= {};
1542
1543     # XXX Add in check here for recycling?
1544
1545     my $engine = $self->engine;
1546
1547     my $blist_loc = $self->get_blist_loc;
1548
1549     # There's no index or blist yet
1550     unless ( $blist_loc ) {
1551         return unless $args->{create};
1552
1553         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1554             engine  => $engine,
1555             key_md5 => $args->{key_md5},
1556         });
1557
1558         $engine->storage->print_at( $self->offset + $self->base_size,
1559             pack( $StP{$engine->byte_size}, $blist->offset ),
1560         );
1561
1562         return $blist;
1563     }
1564
1565     my $sector = $engine->_load_sector( $blist_loc )
1566         or DBM::Deep->_throw_error( "Cannot read sector at $blist_loc in get_bucket_list()" );
1567     my $i = 0;
1568     my $last_sector = undef;
1569     while ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
1570         $blist_loc = $sector->get_entry( ord( substr( $args->{key_md5}, $i++, 1 ) ) );
1571         $last_sector = $sector;
1572         if ( $blist_loc ) {
1573             $sector = $engine->_load_sector( $blist_loc )
1574                 or DBM::Deep->_throw_error( "Cannot read sector at $blist_loc in get_bucket_list()" );
1575         }
1576         else {
1577             $sector = undef;
1578             last;
1579         }
1580     }
1581
1582     # This means we went through the Index sector(s) and found an empty slot
1583     unless ( $sector ) {
1584         return unless $args->{create};
1585
1586         DBM::Deep->_throw_error( "No last_sector when attempting to build a new entry" )
1587             unless $last_sector;
1588
1589         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1590             engine  => $engine,
1591             key_md5 => $args->{key_md5},
1592         });
1593
1594         $last_sector->set_entry( ord( substr( $args->{key_md5}, $i - 1, 1 ) ) => $blist->offset );
1595
1596         return $blist;
1597     }
1598
1599     $sector->find_md5( $args->{key_md5} );
1600
1601     # See whether or not we need to reindex the bucketlist
1602     # Yes, the double-braces are there for a reason. if() doesn't create a redo-able block,
1603     # so we have to create a bare block within the if() for redo-purposes. Patch and idea
1604     # submitted by sprout@cpan.org. -RobK, 2008-01-09
1605     if ( !$sector->has_md5 && $args->{create} && $sector->{idx} == -1 ) {{
1606         my $redo;
1607
1608         my $new_index = DBM::Deep::Engine::Sector::Index->new({
1609             engine => $engine,
1610         });
1611
1612         my %blist_cache;
1613         #XXX q.v. the comments for this function.
1614         foreach my $entry ( $sector->chopped_up ) {
1615             my ($spot, $md5) = @{$entry};
1616             my $idx = ord( substr( $md5, $i, 1 ) );
1617
1618             # XXX This is inefficient
1619             my $blist = $blist_cache{$idx}
1620                 ||= DBM::Deep::Engine::Sector::BucketList->new({
1621                     engine => $engine,
1622                 });
1623
1624             $new_index->set_entry( $idx => $blist->offset );
1625
1626             my $new_spot = $blist->write_at_next_open( $md5 );
1627             $engine->reindex_entry( $spot => $new_spot );
1628         }
1629
1630         # Handle the new item separately.
1631         {
1632             my $idx = ord( substr( $args->{key_md5}, $i, 1 ) );
1633
1634             # If all the previous blist's items have been thrown into one
1635             # blist and the new item belongs in there too, we need
1636             # another index.
1637             if ( keys %blist_cache == 1 and each %blist_cache == $idx ) {
1638                 ++$i, ++$redo;
1639             } else {
1640                 my $blist = $blist_cache{$idx}
1641                     ||= DBM::Deep::Engine::Sector::BucketList->new({
1642                         engine => $engine,
1643                     });
1644
1645                 $new_index->set_entry( $idx => $blist->offset );
1646
1647                 #XXX THIS IS HACKY!
1648                 $blist->find_md5( $args->{key_md5} );
1649                 $blist->write_md5({
1650                     key     => $args->{key},
1651                     key_md5 => $args->{key_md5},
1652                     value   => DBM::Deep::Engine::Sector::Null->new({
1653                         engine => $engine,
1654                         data   => undef,
1655                     }),
1656                 });
1657             }
1658 #            my $blist = $blist_cache{$idx}
1659 #                ||= DBM::Deep::Engine::Sector::BucketList->new({
1660 #                    engine => $engine,
1661 #                });
1662 #
1663 #            $new_index->set_entry( $idx => $blist->offset );
1664 #
1665 #            #XXX THIS IS HACKY!
1666 #            $blist->find_md5( $args->{key_md5} );
1667 #            $blist->write_md5({
1668 #                key     => $args->{key},
1669 #                key_md5 => $args->{key_md5},
1670 #                value   => DBM::Deep::Engine::Sector::Null->new({
1671 #                    engine => $engine,
1672 #                    data   => undef,
1673 #                }),
1674 #            });
1675         }
1676
1677         if ( $last_sector ) {
1678             $last_sector->set_entry(
1679                 ord( substr( $args->{key_md5}, $i - 1, 1 ) ),
1680                 $new_index->offset,
1681             );
1682         } else {
1683             $engine->storage->print_at( $self->offset + $self->base_size,
1684                 pack( $StP{$engine->byte_size}, $new_index->offset ),
1685             );
1686         }
1687
1688         $sector->clear;
1689         $sector->free;
1690
1691         if ( $redo ) {
1692             (undef, $sector) = %blist_cache;
1693             $last_sector = $new_index;
1694             redo;
1695         }
1696
1697         $sector = $blist_cache{ ord( substr( $args->{key_md5}, $i, 1 ) ) };
1698         $sector->find_md5( $args->{key_md5} );
1699     }}
1700
1701     return $sector;
1702 }
1703
1704 sub get_class_offset {
1705     my $self = shift;
1706
1707     my $e = $self->engine;
1708     return unpack(
1709         $StP{$e->byte_size},
1710         $e->storage->read_at(
1711             $self->offset + $self->base_size + 1 * $e->byte_size, $e->byte_size,
1712         ),
1713     );
1714 }
1715
1716 sub get_classname {
1717     my $self = shift;
1718
1719     my $class_offset = $self->get_class_offset;
1720
1721     return unless $class_offset;
1722
1723     return $self->engine->_load_sector( $class_offset )->data;
1724 }
1725
1726 sub data {
1727     my $self = shift;
1728
1729     unless ( $self->engine->cache->{ $self->offset } ) {
1730         my $new_obj = DBM::Deep->new({
1731             type        => $self->type,
1732             base_offset => $self->offset,
1733             staleness   => $self->staleness,
1734             storage     => $self->engine->storage,
1735             engine      => $self->engine,
1736         });
1737
1738         if ( $self->engine->storage->{autobless} ) {
1739             my $classname = $self->get_classname;
1740             if ( defined $classname ) {
1741                 bless $new_obj, $classname;
1742             }
1743         }
1744
1745         $self->engine->cache->{$self->offset} = $new_obj;
1746     }
1747     return $self->engine->cache->{$self->offset};
1748 }
1749
1750 sub free {
1751     my $self = shift;
1752
1753     # We're not ready to be removed yet.
1754     if ( $self->decrement_refcount > 0 ) {
1755         return;
1756     }
1757
1758     # Rebless the object into DBM::Deep::Null.
1759     eval { %{ $self->engine->cache->{ $self->offset } } = (); };
1760     eval { @{ $self->engine->cache->{ $self->offset } } = (); };
1761     bless $self->engine->cache->{ $self->offset }, 'DBM::Deep::Null';
1762     delete $self->engine->cache->{ $self->offset };
1763
1764     my $blist_loc = $self->get_blist_loc;
1765     $self->engine->_load_sector( $blist_loc )->free if $blist_loc;
1766
1767     my $class_loc = $self->get_class_offset;
1768     $self->engine->_load_sector( $class_loc )->free if $class_loc;
1769
1770     $self->SUPER::free();
1771 }
1772
1773 sub increment_refcount {
1774     my $self = shift;
1775
1776     my $refcount = $self->get_refcount;
1777
1778     $refcount++;
1779
1780     $self->write_refcount( $refcount );
1781
1782     return $refcount;
1783 }
1784
1785 sub decrement_refcount {
1786     my $self = shift;
1787
1788     my $refcount = $self->get_refcount;
1789
1790     $refcount--;
1791
1792     $self->write_refcount( $refcount );
1793
1794     return $refcount;
1795 }
1796
1797 sub get_refcount {
1798     my $self = shift;
1799
1800     my $e = $self->engine;
1801     return unpack(
1802         $StP{$e->byte_size},
1803         $e->storage->read_at(
1804             $self->offset + $self->base_size + 2 * $e->byte_size, $e->byte_size,
1805         ),
1806     );
1807 }
1808
1809 sub write_refcount {
1810     my $self = shift;
1811     my ($num) = @_;
1812
1813     my $e = $self->engine;
1814     $e->storage->print_at(
1815         $self->offset + $self->base_size + 2 * $e->byte_size,
1816         pack( $StP{$e->byte_size}, $num ),
1817     );
1818 }
1819
1820 package DBM::Deep::Engine::Sector::BucketList;
1821
1822 our @ISA = qw( DBM::Deep::Engine::Sector );
1823
1824 sub _init {
1825     my $self = shift;
1826
1827     my $engine = $self->engine;
1828
1829     unless ( $self->offset ) {
1830         my $leftover = $self->size - $self->base_size;
1831
1832         $self->{offset} = $engine->_request_blist_sector( $self->size );
1833         $engine->storage->print_at( $self->offset, $engine->SIG_BLIST ); # Sector type
1834         # Skip staleness counter
1835         $engine->storage->print_at( $self->offset + $self->base_size,
1836             chr(0) x $leftover, # Zero-fill the data
1837         );
1838     }
1839
1840     if ( $self->{key_md5} ) {
1841         $self->find_md5;
1842     }
1843
1844     return $self;
1845 }
1846
1847 sub clear {
1848     my $self = shift;
1849     $self->engine->storage->print_at( $self->offset + $self->base_size,
1850         chr(0) x ($self->size - $self->base_size), # Zero-fill the data
1851     );
1852 }
1853
1854 sub size {
1855     my $self = shift;
1856     unless ( $self->{size} ) {
1857         my $e = $self->engine;
1858         # Base + numbuckets * bucketsize
1859         $self->{size} = $self->base_size + $e->max_buckets * $self->bucket_size;
1860     }
1861     return $self->{size};
1862 }
1863
1864 sub free_meth { return '_add_free_blist_sector' }
1865
1866 sub free {
1867     my $self = shift;
1868
1869     my $e = $self->engine;
1870     foreach my $bucket ( $self->chopped_up ) {
1871         my $rest = $bucket->[-1];
1872
1873         # Delete the keysector
1874         my $l = unpack( $StP{$e->byte_size}, substr( $rest, $e->hash_size, $e->byte_size ) );
1875         my $s = $e->_load_sector( $l ); $s->free if $s;
1876
1877         # Delete the HEAD sector
1878         $l = unpack( $StP{$e->byte_size},
1879             substr( $rest,
1880                 $e->hash_size + $e->byte_size,
1881                 $e->byte_size,
1882             ),
1883         );
1884         $s = $e->_load_sector( $l ); $s->free if $s;
1885
1886         foreach my $txn ( 0 .. $e->num_txns - 2 ) {
1887             my $l = unpack( $StP{$e->byte_size},
1888                 substr( $rest,
1889                     $e->hash_size + 2 * $e->byte_size + $txn * ($e->byte_size + $STALE_SIZE),
1890                     $e->byte_size,
1891                 ),
1892             );
1893             my $s = $e->_load_sector( $l ); $s->free if $s;
1894         }
1895     }
1896
1897     $self->SUPER::free();
1898 }
1899
1900 sub bucket_size {
1901     my $self = shift;
1902     unless ( $self->{bucket_size} ) {
1903         my $e = $self->engine;
1904         # Key + head (location) + transactions (location + staleness-counter)
1905         my $location_size = $e->byte_size + $e->byte_size + ($e->num_txns - 1) * ($e->byte_size + $STALE_SIZE);
1906         $self->{bucket_size} = $e->hash_size + $location_size;
1907     }
1908     return $self->{bucket_size};
1909 }
1910
1911 # XXX This is such a poor hack. I need to rethink this code.
1912 sub chopped_up {
1913     my $self = shift;
1914
1915     my $e = $self->engine;
1916
1917     my @buckets;
1918     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1919         my $spot = $self->offset + $self->base_size + $idx * $self->bucket_size;
1920         my $md5 = $e->storage->read_at( $spot, $e->hash_size );
1921
1922         #XXX If we're chopping, why would we ever have the blank_md5?
1923         last if $md5 eq $e->blank_md5;
1924
1925         my $rest = $e->storage->read_at( undef, $self->bucket_size - $e->hash_size );
1926         push @buckets, [ $spot, $md5 . $rest ];
1927     }
1928
1929     return @buckets;
1930 }
1931
1932 sub write_at_next_open {
1933     my $self = shift;
1934     my ($entry) = @_;
1935
1936     #XXX This is such a hack!
1937     $self->{_next_open} = 0 unless exists $self->{_next_open};
1938
1939     my $spot = $self->offset + $self->base_size + $self->{_next_open}++ * $self->bucket_size;
1940     $self->engine->storage->print_at( $spot, $entry );
1941
1942     return $spot;
1943 }
1944
1945 sub has_md5 {
1946     my $self = shift;
1947     unless ( exists $self->{found} ) {
1948         $self->find_md5;
1949     }
1950     return $self->{found};
1951 }
1952
1953 sub find_md5 {
1954     my $self = shift;
1955
1956     $self->{found} = undef;
1957     $self->{idx}   = -1;
1958
1959     if ( @_ ) {
1960         $self->{key_md5} = shift;
1961     }
1962
1963     # If we don't have an MD5, then what are we supposed to do?
1964     unless ( exists $self->{key_md5} ) {
1965         DBM::Deep->_throw_error( "Cannot find_md5 without a key_md5 set" );
1966     }
1967
1968     my $e = $self->engine;
1969     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1970         my $potential = $e->storage->read_at(
1971             $self->offset + $self->base_size + $idx * $self->bucket_size, $e->hash_size,
1972         );
1973
1974         if ( $potential eq $e->blank_md5 ) {
1975             $self->{idx} = $idx;
1976             return;
1977         }
1978
1979         if ( $potential eq $self->{key_md5} ) {
1980             $self->{found} = 1;
1981             $self->{idx} = $idx;
1982             return;
1983         }
1984     }
1985
1986     return;
1987 }
1988
1989 sub write_md5 {
1990     my $self = shift;
1991     my ($args) = @_;
1992
1993     DBM::Deep->_throw_error( "write_md5: no key" ) unless exists $args->{key};
1994     DBM::Deep->_throw_error( "write_md5: no key_md5" ) unless exists $args->{key_md5};
1995     DBM::Deep->_throw_error( "write_md5: no value" ) unless exists $args->{value};
1996
1997     my $engine = $self->engine;
1998
1999     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
2000
2001     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2002     $engine->add_entry( $args->{trans_id}, $spot );
2003
2004     unless ($self->{found}) {
2005         my $key_sector = DBM::Deep::Engine::Sector::Scalar->new({
2006             engine => $engine,
2007             data   => $args->{key},
2008         });
2009
2010         $engine->storage->print_at( $spot,
2011             $args->{key_md5},
2012             pack( $StP{$engine->byte_size}, $key_sector->offset ),
2013         );
2014     }
2015
2016     my $loc = $spot
2017       + $engine->hash_size
2018       + $engine->byte_size;
2019
2020     if ( $args->{trans_id} ) {
2021         $loc += $engine->byte_size + ($args->{trans_id} - 1) * ( $engine->byte_size + $STALE_SIZE );
2022
2023         $engine->storage->print_at( $loc,
2024             pack( $StP{$engine->byte_size}, $args->{value}->offset ),
2025             pack( $StP{$STALE_SIZE}, $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
2026         );
2027     }
2028     else {
2029         $engine->storage->print_at( $loc,
2030             pack( $StP{$engine->byte_size}, $args->{value}->offset ),
2031         );
2032     }
2033 }
2034
2035 sub mark_deleted {
2036     my $self = shift;
2037     my ($args) = @_;
2038     $args ||= {};
2039
2040     my $engine = $self->engine;
2041
2042     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
2043
2044     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2045     $engine->add_entry( $args->{trans_id}, $spot );
2046
2047     my $loc = $spot
2048       + $engine->hash_size
2049       + $engine->byte_size;
2050
2051     if ( $args->{trans_id} ) {
2052         $loc += $engine->byte_size + ($args->{trans_id} - 1) * ( $engine->byte_size + $STALE_SIZE );
2053
2054         $engine->storage->print_at( $loc,
2055             pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
2056             pack( $StP{$STALE_SIZE}, $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
2057         );
2058     }
2059     else {
2060         $engine->storage->print_at( $loc,
2061             pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
2062         );
2063     }
2064
2065 }
2066
2067 sub delete_md5 {
2068     my $self = shift;
2069     my ($args) = @_;
2070
2071     my $engine = $self->engine;
2072     return undef unless $self->{found};
2073
2074     # Save the location so that we can free the data
2075     my $location = $self->get_data_location_for({
2076         allow_head => 0,
2077     });
2078     my $key_sector = $self->get_key_for;
2079
2080     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2081     $engine->storage->print_at( $spot,
2082         $engine->storage->read_at(
2083             $spot + $self->bucket_size,
2084             $self->bucket_size * ( $engine->max_buckets - $self->{idx} - 1 ),
2085         ),
2086         chr(0) x $self->bucket_size,
2087     );
2088
2089     $key_sector->free;
2090
2091     my $data_sector = $self->engine->_load_sector( $location );
2092     my $data = $data_sector->data;
2093     $data_sector->free;
2094
2095     return $data;
2096 }
2097
2098 sub get_data_location_for {
2099     my $self = shift;
2100     my ($args) = @_;
2101     $args ||= {};
2102
2103     $args->{allow_head} = 0 unless exists $args->{allow_head};
2104     $args->{trans_id}   = $self->engine->trans_id unless exists $args->{trans_id};
2105     $args->{idx}        = $self->{idx} unless exists $args->{idx};
2106
2107     my $e = $self->engine;
2108
2109     my $spot = $self->offset + $self->base_size
2110       + $args->{idx} * $self->bucket_size
2111       + $e->hash_size
2112       + $e->byte_size;
2113
2114     if ( $args->{trans_id} ) {
2115         $spot += $e->byte_size + ($args->{trans_id} - 1) * ( $e->byte_size + $STALE_SIZE );
2116     }
2117
2118     my $buffer = $e->storage->read_at(
2119         $spot,
2120         $e->byte_size + $STALE_SIZE,
2121     );
2122     my ($loc, $staleness) = unpack( $StP{$e->byte_size} . ' ' . $StP{$STALE_SIZE}, $buffer );
2123
2124     # XXX Merge the two if-clauses below
2125     if ( $args->{trans_id} ) {
2126         # We have found an entry that is old, so get rid of it
2127         if ( $staleness != (my $s = $e->get_txn_staleness_counter( $args->{trans_id} ) ) ) {
2128             $e->storage->print_at(
2129                 $spot,
2130                 pack( $StP{$e->byte_size} . ' ' . $StP{$STALE_SIZE}, (0) x 2 ),
2131             );
2132             $loc = 0;
2133         }
2134     }
2135
2136     # If we're in a transaction and we never wrote to this location, try the
2137     # HEAD instead.
2138     if ( $args->{trans_id} && !$loc && $args->{allow_head} ) {
2139         return $self->get_data_location_for({
2140             trans_id   => 0,
2141             allow_head => 1,
2142             idx        => $args->{idx},
2143         });
2144     }
2145
2146     return $loc <= 1 ? 0 : $loc;
2147 }
2148
2149 sub get_data_for {
2150     my $self = shift;
2151     my ($args) = @_;
2152     $args ||= {};
2153
2154     return unless $self->{found};
2155     my $location = $self->get_data_location_for({
2156         allow_head => $args->{allow_head},
2157     });
2158     return $self->engine->_load_sector( $location );
2159 }
2160
2161 sub get_key_for {
2162     my $self = shift;
2163     my ($idx) = @_;
2164     $idx = $self->{idx} unless defined $idx;
2165
2166     if ( $idx >= $self->engine->max_buckets ) {
2167         DBM::Deep->_throw_error( "get_key_for(): Attempting to retrieve $idx" );
2168     }
2169
2170     my $location = $self->engine->storage->read_at(
2171         $self->offset + $self->base_size + $idx * $self->bucket_size + $self->engine->hash_size,
2172         $self->engine->byte_size,
2173     );
2174     $location = unpack( $StP{$self->engine->byte_size}, $location );
2175     DBM::Deep->_throw_error( "get_key_for: No location?" ) unless $location;
2176
2177     return $self->engine->_load_sector( $location );
2178 }
2179
2180 package DBM::Deep::Engine::Sector::Index;
2181
2182 our @ISA = qw( DBM::Deep::Engine::Sector );
2183
2184 sub _init {
2185     my $self = shift;
2186
2187     my $engine = $self->engine;
2188
2189     unless ( $self->offset ) {
2190         my $leftover = $self->size - $self->base_size;
2191
2192         $self->{offset} = $engine->_request_index_sector( $self->size );
2193         $engine->storage->print_at( $self->offset, $engine->SIG_INDEX ); # Sector type
2194         # Skip staleness counter
2195         $engine->storage->print_at( $self->offset + $self->base_size,
2196             chr(0) x $leftover, # Zero-fill the rest
2197         );
2198     }
2199
2200     return $self;
2201 }
2202
2203 #XXX Change here
2204 sub size {
2205     my $self = shift;
2206     unless ( $self->{size} ) {
2207         my $e = $self->engine;
2208         $self->{size} = $self->base_size + $e->byte_size * $e->hash_chars;
2209     }
2210     return $self->{size};
2211 }
2212
2213 sub free_meth { return '_add_free_index_sector' }
2214
2215 sub free {
2216     my $self = shift;
2217     my $e = $self->engine;
2218
2219     for my $i ( 0 .. $e->hash_chars - 1 ) {
2220         my $l = $self->get_entry( $i ) or next;
2221         $e->_load_sector( $l )->free;
2222     }
2223
2224     $self->SUPER::free();
2225 }
2226
2227 sub _loc_for {
2228     my $self = shift;
2229     my ($idx) = @_;
2230     return $self->offset + $self->base_size + $idx * $self->engine->byte_size;
2231 }
2232
2233 sub get_entry {
2234     my $self = shift;
2235     my ($idx) = @_;
2236
2237     my $e = $self->engine;
2238
2239     DBM::Deep->_throw_error( "get_entry: Out of range ($idx)" )
2240         if $idx < 0 || $idx >= $e->hash_chars;
2241
2242     return unpack(
2243         $StP{$e->byte_size},
2244         $e->storage->read_at( $self->_loc_for( $idx ), $e->byte_size ),
2245     );
2246 }
2247
2248 sub set_entry {
2249     my $self = shift;
2250     my ($idx, $loc) = @_;
2251
2252     my $e = $self->engine;
2253
2254     DBM::Deep->_throw_error( "set_entry: Out of range ($idx)" )
2255         if $idx < 0 || $idx >= $e->hash_chars;
2256
2257     $self->engine->storage->print_at(
2258         $self->_loc_for( $idx ),
2259         pack( $StP{$e->byte_size}, $loc ),
2260     );
2261 }
2262
2263 # This was copied from MARCEL's Class::Null. However, I couldn't use it because
2264 # I need an undef value, not an implementation of the Null Class pattern.
2265 package DBM::Deep::Null;
2266
2267 use overload
2268     'bool'   => sub { undef },
2269     '""'     => sub { undef },
2270     '0+'     => sub { undef },
2271     fallback => 1,
2272     nomethod => 'AUTOLOAD';
2273
2274 sub AUTOLOAD { return; }
2275
2276 1;
2277 __END__