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