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.0012);
   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         foreach my $orig_loc ( keys %{ $locs } ) {
 632             if ( $orig_loc == $old_loc ) {
 633                 delete $locs->{orig_loc};
 634                 $locs->{$new_loc} = undef;
 635                 next TRANS;
 636             }
 637         }
 638     }
 639 }
 640
 641 sub clear_entries {
 642     my $self = shift;
 643     delete $self->{entries}{$self->trans_id};
 644 }
 645
 646 ################################################################################
 647
 648 {
 649     my $header_fixed = length( SIG_FILE ) + 1 + 4 + 4;
 650     my $this_file_version = 3;
 651
 652     sub _write_file_header {
 653         my $self = shift;
 654
 655         my $nt = $self->num_txns;
 656         my $bl = $self->txn_bitfield_len;
 657
 658         my $header_var = 1 + 1 + 1 + 1 + $bl + $STALE_SIZE * ($nt - 1) + 3 * $self->byte_size;
 659
 660         my $loc = $self->storage->request_space( $header_fixed + $header_var );
 661
 662         $self->storage->print_at( $loc,
 663             SIG_FILE,
 664             SIG_HEADER,
 665             pack('N', $this_file_version), # At this point, we're at 9 bytes
 666             pack('N', $header_var),        # header size
 667             # --- Above is $header_fixed. Below is $header_var
 668             pack('C', $self->byte_size),
 669
 670             # These shenanigans are to allow a 256 within a C
 671             pack('C', $self->max_buckets - 1),
 672             pack('C', $self->data_sector_size - 1),
 673
 674             pack('C', $nt),
 675             pack('C' . $bl, 0 ),                           # Transaction activeness bitfield
 676             pack($StP{$STALE_SIZE}.($nt-1), 0 x ($nt-1) ), # Transaction staleness counters
 677             pack($StP{$self->byte_size}, 0), # Start of free chain (blist size)
 678             pack($StP{$self->byte_size}, 0), # Start of free chain (data size)
 679             pack($StP{$self->byte_size}, 0), # Start of free chain (index size)
 680         );
 681
 682         #XXX Set these less fragilely
 683         $self->set_trans_loc( $header_fixed + 4 );
 684         $self->set_chains_loc( $header_fixed + 4 + $bl + $STALE_SIZE * ($nt-1) );
 685
 686         return;
 687     }
 688
 689     sub _read_file_header {
 690         my $self = shift;
 691
 692         my $buffer = $self->storage->read_at( 0, $header_fixed );
 693         return unless length($buffer);
 694
 695         my ($file_signature, $sig_header, $file_version, $size) = unpack(
 696             'A4 A N N', $buffer
 697         );
 698
 699         unless ( $file_signature eq SIG_FILE ) {
 700             $self->storage->close;
 701             DBM::Deep->_throw_error( "Signature not found -- file is not a Deep DB" );
 702         }
 703
 704         unless ( $sig_header eq SIG_HEADER ) {
 705             $self->storage->close;
 706             DBM::Deep->_throw_error( "Pre-1.00 file version found" );
 707         }
 708
 709         unless ( $file_version == $this_file_version ) {
 710             $self->storage->close;
 711             DBM::Deep->_throw_error(
 712                 "Wrong file version found - " .  $file_version .
 713                 " - expected " . $this_file_version
 714             );
 715         }
 716
 717         my $buffer2 = $self->storage->read_at( undef, $size );
 718         my @values = unpack( 'C C C C', $buffer2 );
 719
 720         if ( @values != 4 || grep { !defined } @values ) {
 721             $self->storage->close;
 722             DBM::Deep->_throw_error("Corrupted file - bad header");
 723         }
 724
 725         #XXX Add warnings if values weren't set right
 726         @{$self}{qw(byte_size max_buckets data_sector_size num_txns)} = @values;
 727
 728         # These shenangians are to allow a 256 within a C
 729         $self->{max_buckets} += 1;
 730         $self->{data_sector_size} += 1;
 731
 732         my $bl = $self->txn_bitfield_len;
 733
 734         my $header_var = scalar(@values) + $bl + $STALE_SIZE * ($self->num_txns - 1) + 3 * $self->byte_size;
 735         unless ( $size == $header_var ) {
 736             $self->storage->close;
 737             DBM::Deep->_throw_error( "Unexpected size found ($size <-> $header_var)." );
 738         }
 739
 740         $self->set_trans_loc( $header_fixed + scalar(@values) );
 741         $self->set_chains_loc( $header_fixed + scalar(@values) + $bl + $STALE_SIZE * ($self->num_txns - 1) );
 742
 743         return length($buffer) + length($buffer2);
 744     }
 745 }
 746
 747 sub _load_sector {
 748     my $self = shift;
 749     my ($offset) = @_;
 750
 751     # Add a catch for offset of 0 or 1
 752     return if !$offset || $offset <= 1;
 753
 754     my $type = $self->storage->read_at( $offset, 1 );
 755     return if $type eq chr(0);
 756
 757     if ( $type eq $self->SIG_ARRAY || $type eq $self->SIG_HASH ) {
 758         return DBM::Deep::Engine::Sector::Reference->new({
 759             engine => $self,
 760             type   => $type,
 761             offset => $offset,
 762         });
 763     }
 764     # XXX Don't we need key_md5 here?
 765     elsif ( $type eq $self->SIG_BLIST ) {
 766         return DBM::Deep::Engine::Sector::BucketList->new({
 767             engine => $self,
 768             type   => $type,
 769             offset => $offset,
 770         });
 771     }
 772     elsif ( $type eq $self->SIG_INDEX ) {
 773         return DBM::Deep::Engine::Sector::Index->new({
 774             engine => $self,
 775             type   => $type,
 776             offset => $offset,
 777         });
 778     }
 779     elsif ( $type eq $self->SIG_NULL ) {
 780         return DBM::Deep::Engine::Sector::Null->new({
 781             engine => $self,
 782             type   => $type,
 783             offset => $offset,
 784         });
 785     }
 786     elsif ( $type eq $self->SIG_DATA ) {
 787         return DBM::Deep::Engine::Sector::Scalar->new({
 788             engine => $self,
 789             type   => $type,
 790             offset => $offset,
 791         });
 792     }
 793     # This was deleted from under us, so just return and let the caller figure it out.
 794     elsif ( $type eq $self->SIG_FREE ) {
 795         return;
 796     }
 797
 798     DBM::Deep->_throw_error( "'$offset': Don't know what to do with type '$type'" );
 799 }
 800
 801 sub _apply_digest {
 802     my $self = shift;
 803     return $self->{digest}->(@_);
 804 }
 805
 806 sub _add_free_blist_sector { shift->_add_free_sector( 0, @_ ) }
 807 sub _add_free_data_sector { shift->_add_free_sector( 1, @_ ) }
 808 sub _add_free_index_sector { shift->_add_free_sector( 2, @_ ) }
 809
 810 sub _add_free_sector {
 811     my $self = shift;
 812     my ($multiple, $offset, $size) = @_;
 813
 814     my $chains_offset = $multiple * $self->byte_size;
 815
 816     my $storage = $self->storage;
 817
 818     # Increment staleness.
 819     # XXX Can this increment+modulo be done by "&= 0x1" ?
 820     my $staleness = unpack( $StP{$STALE_SIZE}, $storage->read_at( $offset + SIG_SIZE, $STALE_SIZE ) );
 821     $staleness = ($staleness + 1 ) % ( 2 ** ( 8 * $STALE_SIZE ) );
 822     $storage->print_at( $offset + SIG_SIZE, pack( $StP{$STALE_SIZE}, $staleness ) );
 823
 824     my $old_head = $storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 825
 826     $storage->print_at( $self->chains_loc + $chains_offset,
 827         pack( $StP{$self->byte_size}, $offset ),
 828     );
 829
 830     # Record the old head in the new sector after the signature and staleness counter
 831     $storage->print_at( $offset + SIG_SIZE + $STALE_SIZE, $old_head );
 832 }
 833
 834 sub _request_blist_sector { shift->_request_sector( 0, @_ ) }
 835 sub _request_data_sector { shift->_request_sector( 1, @_ ) }
 836 sub _request_index_sector { shift->_request_sector( 2, @_ ) }
 837
 838 sub _request_sector {
 839     my $self = shift;
 840     my ($multiple, $size) = @_;
 841
 842     my $chains_offset = $multiple * $self->byte_size;
 843
 844     my $old_head = $self->storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 845     my $loc = unpack( $StP{$self->byte_size}, $old_head );
 846
 847     # We don't have any free sectors of the right size, so allocate a new one.
 848     unless ( $loc ) {
 849         my $offset = $self->storage->request_space( $size );
 850
 851         # Zero out the new sector. This also guarantees correct increases
 852         # in the filesize.
 853         $self->storage->print_at( $offset, chr(0) x $size );
 854
 855         return $offset;
 856     }
 857
 858     # Read the new head after the signature and the staleness counter
 859     my $new_head = $self->storage->read_at( $loc + SIG_SIZE + $STALE_SIZE, $self->byte_size );
 860     $self->storage->print_at( $self->chains_loc + $chains_offset, $new_head );
 861     $self->storage->print_at(
 862         $loc + SIG_SIZE + $STALE_SIZE,
 863         pack( $StP{$self->byte_size}, 0 ),
 864     );
 865
 866     return $loc;
 867 }
 868
 869 ################################################################################
 870
 871 sub storage     { $_[0]{storage} }
 872 sub byte_size   { $_[0]{byte_size} }
 873 sub hash_size   { $_[0]{hash_size} }
 874 sub hash_chars  { $_[0]{hash_chars} }
 875 sub num_txns    { $_[0]{num_txns} }
 876 sub max_buckets { $_[0]{max_buckets} }
 877 sub blank_md5   { chr(0) x $_[0]->hash_size }
 878 sub data_sector_size { $_[0]{data_sector_size} }
 879
 880 # This is a calculated value
 881 sub txn_bitfield_len {
 882     my $self = shift;
 883     unless ( exists $self->{txn_bitfield_len} ) {
 884         my $temp = ($self->num_txns) / 8;
 885         if ( $temp > int( $temp ) ) {
 886             $temp = int( $temp ) + 1;
 887         }
 888         $self->{txn_bitfield_len} = $temp;
 889     }
 890     return $self->{txn_bitfield_len};
 891 }
 892
 893 sub trans_id     { $_[0]{trans_id} }
 894 sub set_trans_id { $_[0]{trans_id} = $_[1] }
 895
 896 sub trans_loc     { $_[0]{trans_loc} }
 897 sub set_trans_loc { $_[0]{trans_loc} = $_[1] }
 898
 899 sub chains_loc     { $_[0]{chains_loc} }
 900 sub set_chains_loc { $_[0]{chains_loc} = $_[1] }
 901
 902 sub cache       { $_[0]{cache} ||= {} }
 903 sub clear_cache { %{$_[0]->cache} = () }
 904
 905 sub _dump_file {
 906     my $self = shift;
 907
 908     # Read the header
 909     my $spot = $self->_read_file_header();
 910
 911     my %types = (
 912         0 => 'B',
 913         1 => 'D',
 914         2 => 'I',
 915     );
 916
 917     my %sizes = (
 918         'D' => $self->data_sector_size,
 919         'B' => DBM::Deep::Engine::Sector::BucketList->new({engine=>$self,offset=>1})->size,
 920         'I' => DBM::Deep::Engine::Sector::Index->new({engine=>$self,offset=>1})->size,
 921     );
 922
 923     my $return = "";
 924
 925     # Header values
 926     $return .= "NumTxns: " . $self->num_txns . $/;
 927
 928     # Read the free sector chains
 929     my %sectors;
 930     foreach my $multiple ( 0 .. 2 ) {
 931         $return .= "Chains($types{$multiple}):";
 932         my $old_loc = $self->chains_loc + $multiple * $self->byte_size;
 933         while ( 1 ) {
 934             my $loc = unpack(
 935                 $StP{$self->byte_size},
 936                 $self->storage->read_at( $old_loc, $self->byte_size ),
 937             );
 938
 939             # We're now out of free sectors of this kind.
 940             unless ( $loc ) {
 941                 last;
 942             }
 943
 944             $sectors{ $types{$multiple} }{ $loc } = undef;
 945             $old_loc = $loc + SIG_SIZE + $STALE_SIZE;
 946             $return .= " $loc";
 947         }
 948         $return .= $/;
 949     }
 950
 951     SECTOR:
 952     while ( $spot < $self->storage->{end} ) {
 953         # Read each sector in order.
 954         my $sector = $self->_load_sector( $spot );
 955         if ( !$sector ) {
 956             # Find it in the free-sectors that were found already
 957             foreach my $type ( keys %sectors ) {
 958                 if ( exists $sectors{$type}{$spot} ) {
 959                     my $size = $sizes{$type};
 960                     $return .= sprintf "%08d: %s %04d\n", $spot, 'F' . $type, $size;
 961                     $spot += $size;
 962                     next SECTOR;
 963                 }
 964             }
 965
 966             die "********\n$return\nDidn't find free sector for $spot in chains\n********\n";
 967         }
 968         else {
 969             $return .= sprintf "%08d: %s  %04d", $spot, $sector->type, $sector->size;
 970             if ( $sector->type eq 'D' ) {
 971                 $return .= ' ' . $sector->data;
 972             }
 973             elsif ( $sector->type eq 'A' || $sector->type eq 'H' ) {
 974                 $return .= ' REF: ' . $sector->get_refcount;
 975             }
 976             elsif ( $sector->type eq 'B' ) {
 977                 foreach my $bucket ( $sector->chopped_up ) {
 978                     $return .= "\n    ";
 979                     $return .= sprintf "%08d", unpack($StP{$self->byte_size},
 980                         substr( $bucket->[-1], $self->hash_size, $self->byte_size),
 981                     );
 982                     my $l = unpack( $StP{$self->byte_size},
 983                         substr( $bucket->[-1],
 984                             $self->hash_size + $self->byte_size,
 985                             $self->byte_size,
 986                         ),
 987                     );
 988                     $return .= sprintf " %08d", $l;
 989                     foreach my $txn ( 0 .. $self->num_txns - 2 ) {
 990                         my $l = unpack( $StP{$self->byte_size},
 991                             substr( $bucket->[-1],
 992                                 $self->hash_size + 2 * $self->byte_size + $txn * ($self->byte_size + $STALE_SIZE),
 993                                 $self->byte_size,
 994                             ),
 995                         );
 996                         $return .= sprintf " %08d", $l;
 997                     }
 998                 }
 999             }
1000             $return .= $/;
1001
1002             $spot += $sector->size;
1003         }
1004     }
1005
1006     return $return;
1007 }
1008
1009 ################################################################################
1010
1011 package DBM::Deep::Iterator;
1012
1013 sub new {
1014     my $class = shift;
1015     my ($args) = @_;
1016
1017     my $self = bless {
1018         breadcrumbs => [],
1019         engine      => $args->{engine},
1020         base_offset => $args->{base_offset},
1021     }, $class;
1022
1023     Scalar::Util::weaken( $self->{engine} );
1024
1025     return $self;
1026 }
1027
1028 sub reset { $_[0]{breadcrumbs} = [] }
1029
1030 sub get_sector_iterator {
1031     my $self = shift;
1032     my ($loc) = @_;
1033
1034     my $sector = $self->{engine}->_load_sector( $loc )
1035         or return;
1036
1037     if ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
1038         return DBM::Deep::Iterator::Index->new({
1039             iterator => $self,
1040             sector   => $sector,
1041         });
1042     }
1043     elsif ( $sector->isa( 'DBM::Deep::Engine::Sector::BucketList' ) ) {
1044         return DBM::Deep::Iterator::BucketList->new({
1045             iterator => $self,
1046             sector   => $sector,
1047         });
1048     }
1049
1050     DBM::Deep->_throw_error( "get_sector_iterator(): Why did $loc make a $sector?" );
1051 }
1052
1053 sub get_next_key {
1054     my $self = shift;
1055     my ($obj) = @_;
1056
1057     my $crumbs = $self->{breadcrumbs};
1058     my $e = $self->{engine};
1059
1060     unless ( @$crumbs ) {
1061         # This will be a Reference sector
1062         my $sector = $e->_load_sector( $self->{base_offset} )
1063             # If no sector is found, thist must have been deleted from under us.
1064             or return;
1065
1066         if ( $sector->staleness != $obj->_staleness ) {
1067             return;
1068         }
1069
1070         my $loc = $sector->get_blist_loc
1071             or return;
1072
1073         push @$crumbs, $self->get_sector_iterator( $loc );
1074     }
1075
1076     FIND_NEXT_KEY: {
1077         # We're at the end.
1078         unless ( @$crumbs ) {
1079             $self->reset;
1080             return;
1081         }
1082
1083         my $iterator = $crumbs->[-1];
1084
1085         # This level is done.
1086         if ( $iterator->at_end ) {
1087             pop @$crumbs;
1088             redo FIND_NEXT_KEY;
1089         }
1090
1091         if ( $iterator->isa( 'DBM::Deep::Iterator::Index' ) ) {
1092             # If we don't have any more, it will be caught at the
1093             # prior check.
1094             if ( my $next = $iterator->get_next_iterator ) {
1095                 push @$crumbs, $next;
1096             }
1097             redo FIND_NEXT_KEY;
1098         }
1099
1100         unless ( $iterator->isa( 'DBM::Deep::Iterator::BucketList' ) ) {
1101             DBM::Deep->_throw_error(
1102                 "Should have a bucketlist iterator here - instead have $iterator"
1103             );
1104         }
1105
1106         # At this point, we have a BucketList iterator
1107         my $key = $iterator->get_next_key;
1108         if ( defined $key ) {
1109             return $key;
1110         }
1111         #XXX else { $iterator->set_to_end() } ?
1112
1113         # We hit the end of the bucketlist iterator, so redo
1114         redo FIND_NEXT_KEY;
1115     }
1116
1117     DBM::Deep->_throw_error( "get_next_key(): How did we get here?" );
1118 }
1119
1120 package DBM::Deep::Iterator::Index;
1121
1122 sub new {
1123     my $self = bless $_[1] => $_[0];
1124     $self->{curr_index} = 0;
1125     return $self;
1126 }
1127
1128 sub at_end {
1129     my $self = shift;
1130     return $self->{curr_index} >= $self->{iterator}{engine}->hash_chars;
1131 }
1132
1133 sub get_next_iterator {
1134     my $self = shift;
1135
1136     my $loc;
1137     while ( !$loc ) {
1138         return if $self->at_end;
1139         $loc = $self->{sector}->get_entry( $self->{curr_index}++ );
1140     }
1141
1142     return $self->{iterator}->get_sector_iterator( $loc );
1143 }
1144
1145 package DBM::Deep::Iterator::BucketList;
1146
1147 sub new {
1148     my $self = bless $_[1] => $_[0];
1149     $self->{curr_index} = 0;
1150     return $self;
1151 }
1152
1153 sub at_end {
1154     my $self = shift;
1155     return $self->{curr_index} >= $self->{iterator}{engine}->max_buckets;
1156 }
1157
1158 sub get_next_key {
1159     my $self = shift;
1160
1161     return if $self->at_end;
1162
1163     my $idx = $self->{curr_index}++;
1164
1165     my $data_loc = $self->{sector}->get_data_location_for({
1166         allow_head => 1,
1167         idx        => $idx,
1168     }) or return;
1169
1170     #XXX Do we want to add corruption checks here?
1171     return $self->{sector}->get_key_for( $idx )->data;
1172 }
1173
1174 package DBM::Deep::Engine::Sector;
1175
1176 sub new {
1177     my $self = bless $_[1], $_[0];
1178     Scalar::Util::weaken( $self->{engine} );
1179     $self->_init;
1180     return $self;
1181 }
1182
1183 #sub _init {}
1184 #sub clone { DBM::Deep->_throw_error( "Must be implemented in the child class" ); }
1185
1186 sub engine { $_[0]{engine} }
1187 sub offset { $_[0]{offset} }
1188 sub type   { $_[0]{type} }
1189
1190 sub base_size {
1191    my $self = shift;
1192    return $self->engine->SIG_SIZE + $STALE_SIZE;
1193 }
1194
1195 sub free {
1196     my $self = shift;
1197
1198     my $e = $self->engine;
1199
1200     $e->storage->print_at( $self->offset, $e->SIG_FREE );
1201     # Skip staleness counter
1202     $e->storage->print_at( $self->offset + $self->base_size,
1203         chr(0) x ($self->size - $self->base_size),
1204     );
1205
1206     my $free_meth = $self->free_meth;
1207     $e->$free_meth( $self->offset, $self->size );
1208
1209     return;
1210 }
1211
1212 package DBM::Deep::Engine::Sector::Data;
1213
1214 our @ISA = qw( DBM::Deep::Engine::Sector );
1215
1216 # This is in bytes
1217 sub size { $_[0]{engine}->data_sector_size }
1218 sub free_meth { return '_add_free_data_sector' }
1219
1220 sub clone {
1221     my $self = shift;
1222     return ref($self)->new({
1223         engine => $self->engine,
1224         type   => $self->type,
1225         data   => $self->data,
1226     });
1227 }
1228
1229 package DBM::Deep::Engine::Sector::Scalar;
1230
1231 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1232
1233 sub free {
1234     my $self = shift;
1235
1236     my $chain_loc = $self->chain_loc;
1237
1238     $self->SUPER::free();
1239
1240     if ( $chain_loc ) {
1241         $self->engine->_load_sector( $chain_loc )->free;
1242     }
1243
1244     return;
1245 }
1246
1247 sub type { $_[0]{engine}->SIG_DATA }
1248 sub _init {
1249     my $self = shift;
1250
1251     my $engine = $self->engine;
1252
1253     unless ( $self->offset ) {
1254         my $data_section = $self->size - $self->base_size - $engine->byte_size - 1;
1255
1256         $self->{offset} = $engine->_request_data_sector( $self->size );
1257
1258         my $data = delete $self->{data};
1259         my $dlen = length $data;
1260         my $continue = 1;
1261         my $curr_offset = $self->offset;
1262         while ( $continue ) {
1263
1264             my $next_offset = 0;
1265
1266             my ($leftover, $this_len, $chunk);
1267             if ( $dlen > $data_section ) {
1268                 $leftover = 0;
1269                 $this_len = $data_section;
1270                 $chunk = substr( $data, 0, $this_len );
1271
1272                 $dlen -= $data_section;
1273                 $next_offset = $engine->_request_data_sector( $self->size );
1274                 $data = substr( $data, $this_len );
1275             }
1276             else {
1277                 $leftover = $data_section - $dlen;
1278                 $this_len = $dlen;
1279                 $chunk = $data;
1280
1281                 $continue = 0;
1282             }
1283
1284             $engine->storage->print_at( $curr_offset, $self->type ); # Sector type
1285             # Skip staleness
1286             $engine->storage->print_at( $curr_offset + $self->base_size,
1287                 pack( $StP{$engine->byte_size}, $next_offset ),  # Chain loc
1288                 pack( $StP{1}, $this_len ),                      # Data length
1289                 $chunk,                                          # Data to be stored in this sector
1290                 chr(0) x $leftover,                              # Zero-fill the rest
1291             );
1292
1293             $curr_offset = $next_offset;
1294         }
1295
1296         return;
1297     }
1298 }
1299
1300 sub data_length {
1301     my $self = shift;
1302
1303     my $buffer = $self->engine->storage->read_at(
1304         $self->offset + $self->base_size + $self->engine->byte_size, 1
1305     );
1306
1307     return unpack( $StP{1}, $buffer );
1308 }
1309
1310 sub chain_loc {
1311     my $self = shift;
1312     return unpack(
1313         $StP{$self->engine->byte_size},
1314         $self->engine->storage->read_at(
1315             $self->offset + $self->base_size,
1316             $self->engine->byte_size,
1317         ),
1318     );
1319 }
1320
1321 sub data {
1322     my $self = shift;
1323 #    my ($args) = @_;
1324 #    $args ||= {};
1325
1326     my $data;
1327     while ( 1 ) {
1328         my $chain_loc = $self->chain_loc;
1329
1330         $data .= $self->engine->storage->read_at(
1331             $self->offset + $self->base_size + $self->engine->byte_size + 1, $self->data_length,
1332         );
1333
1334         last unless $chain_loc;
1335
1336         $self = $self->engine->_load_sector( $chain_loc );
1337     }
1338
1339     return $data;
1340 }
1341
1342 package DBM::Deep::Engine::Sector::Null;
1343
1344 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1345
1346 sub type { $_[0]{engine}->SIG_NULL }
1347 sub data_length { 0 }
1348 sub data { return }
1349
1350 sub _init {
1351     my $self = shift;
1352
1353     my $engine = $self->engine;
1354
1355     unless ( $self->offset ) {
1356         my $leftover = $self->size - $self->base_size - 1 * $engine->byte_size - 1;
1357
1358         $self->{offset} = $engine->_request_data_sector( $self->size );
1359         $engine->storage->print_at( $self->offset, $self->type ); # Sector type
1360         # Skip staleness counter
1361         $engine->storage->print_at( $self->offset + $self->base_size,
1362             pack( $StP{$engine->byte_size}, 0 ),  # Chain loc
1363             pack( $StP{1}, $self->data_length ),  # Data length
1364             chr(0) x $leftover,                   # Zero-fill the rest
1365         );
1366
1367         return;
1368     }
1369 }
1370
1371 package DBM::Deep::Engine::Sector::Reference;
1372
1373 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1374
1375 sub _init {
1376     my $self = shift;
1377
1378     my $e = $self->engine;
1379
1380     unless ( $self->offset ) {
1381         my $classname = Scalar::Util::blessed( delete $self->{data} );
1382         my $leftover = $self->size - $self->base_size - 3 * $e->byte_size;
1383
1384         my $class_offset = 0;
1385         if ( defined $classname ) {
1386             my $class_sector = DBM::Deep::Engine::Sector::Scalar->new({
1387                 engine => $e,
1388                 data   => $classname,
1389             });
1390             $class_offset = $class_sector->offset;
1391         }
1392
1393         $self->{offset} = $e->_request_data_sector( $self->size );
1394         $e->storage->print_at( $self->offset, $self->type ); # Sector type
1395         # Skip staleness counter
1396         $e->storage->print_at( $self->offset + $self->base_size,
1397             pack( $StP{$e->byte_size}, 0 ),             # Index/BList loc
1398             pack( $StP{$e->byte_size}, $class_offset ), # Classname loc
1399             pack( $StP{$e->byte_size}, 1 ),             # Initial refcount
1400             chr(0) x $leftover,                         # Zero-fill the rest
1401         );
1402     }
1403     else {
1404         $self->{type} = $e->storage->read_at( $self->offset, 1 );
1405     }
1406
1407     $self->{staleness} = unpack(
1408         $StP{$STALE_SIZE},
1409         $e->storage->read_at( $self->offset + $e->SIG_SIZE, $STALE_SIZE ),
1410     );
1411
1412     return;
1413 }
1414
1415 sub staleness { $_[0]{staleness} }
1416
1417 sub get_data_location_for {
1418     my $self = shift;
1419     my ($args) = @_;
1420
1421     # Assume that the head is not allowed unless otherwise specified.
1422     $args->{allow_head} = 0 unless exists $args->{allow_head};
1423
1424     # Assume we don't create a new blist location unless otherwise specified.
1425     $args->{create} = 0 unless exists $args->{create};
1426
1427     my $blist = $self->get_bucket_list({
1428         key_md5 => $args->{key_md5},
1429         key => $args->{key},
1430         create  => $args->{create},
1431     });
1432     return unless $blist && $blist->{found};
1433
1434     # At this point, $blist knows where the md5 is. What it -doesn't- know yet
1435     # is whether or not this transaction has this key. That's part of the next
1436     # function call.
1437     my $location = $blist->get_data_location_for({
1438         allow_head => $args->{allow_head},
1439     }) or return;
1440
1441     return $location;
1442 }
1443
1444 sub get_data_for {
1445     my $self = shift;
1446     my ($args) = @_;
1447
1448     my $location = $self->get_data_location_for( $args )
1449         or return;
1450
1451     return $self->engine->_load_sector( $location );
1452 }
1453
1454 sub write_data {
1455     my $self = shift;
1456     my ($args) = @_;
1457
1458     my $blist = $self->get_bucket_list({
1459         key_md5 => $args->{key_md5},
1460         key => $args->{key},
1461         create  => 1,
1462     }) or DBM::Deep->_throw_error( "How did write_data fail (no blist)?!" );
1463
1464     # Handle any transactional bookkeeping.
1465     if ( $self->engine->trans_id ) {
1466         if ( ! $blist->has_md5 ) {
1467             $blist->mark_deleted({
1468                 trans_id => 0,
1469             });
1470         }
1471     }
1472     else {
1473         my @trans_ids = $self->engine->get_running_txn_ids;
1474         if ( $blist->has_md5 ) {
1475             if ( @trans_ids ) {
1476                 my $old_value = $blist->get_data_for;
1477                 foreach my $other_trans_id ( @trans_ids ) {
1478                     next if $blist->get_data_location_for({
1479                         trans_id   => $other_trans_id,
1480                         allow_head => 0,
1481                     });
1482                     $blist->write_md5({
1483                         trans_id => $other_trans_id,
1484                         key      => $args->{key},
1485                         key_md5  => $args->{key_md5},
1486                         value    => $old_value->clone,
1487                     });
1488                 }
1489             }
1490         }
1491         else {
1492             if ( @trans_ids ) {
1493                 foreach my $other_trans_id ( @trans_ids ) {
1494                     #XXX This doesn't seem to possible to ever happen . . .
1495                     next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1496                     $blist->mark_deleted({
1497                         trans_id => $other_trans_id,
1498                     });
1499                 }
1500             }
1501         }
1502     }
1503
1504     #XXX Is this safe to do transactionally?
1505     # Free the place we're about to write to.
1506     if ( $blist->get_data_location_for({ allow_head => 0 }) ) {
1507         $blist->get_data_for({ allow_head => 0 })->free;
1508     }
1509
1510     $blist->write_md5({
1511         key      => $args->{key},
1512         key_md5  => $args->{key_md5},
1513         value    => $args->{value},
1514     });
1515 }
1516
1517 sub delete_key {
1518     my $self = shift;
1519     my ($args) = @_;
1520
1521     # XXX What should happen if this fails?
1522     my $blist = $self->get_bucket_list({
1523         key_md5 => $args->{key_md5},
1524     }) or DBM::Deep->_throw_error( "How did delete_key fail (no blist)?!" );
1525
1526     # Save the location so that we can free the data
1527     my $location = $blist->get_data_location_for({
1528         allow_head => 0,
1529     });
1530     my $old_value = $location && $self->engine->_load_sector( $location );
1531
1532     my @trans_ids = $self->engine->get_running_txn_ids;
1533
1534     # If we're the HEAD and there are running txns, then we need to clone this value to the other
1535     # transactions to preserve Isolation.
1536     if ( $self->engine->trans_id == 0 ) {
1537         if ( @trans_ids ) {
1538             foreach my $other_trans_id ( @trans_ids ) {
1539                 next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1540                 $blist->write_md5({
1541                     trans_id => $other_trans_id,
1542                     key      => $args->{key},
1543                     key_md5  => $args->{key_md5},
1544                     value    => $old_value->clone,
1545                 });
1546             }
1547         }
1548     }
1549
1550     my $data;
1551     if ( @trans_ids ) {
1552         $blist->mark_deleted( $args );
1553
1554         if ( $old_value ) {
1555             $data = $old_value->data({ export => 1 });
1556             $old_value->free;
1557         }
1558     }
1559     else {
1560         $data = $blist->delete_md5( $args );
1561     }
1562
1563     return $data;
1564 }
1565
1566 sub get_blist_loc {
1567     my $self = shift;
1568
1569     my $e = $self->engine;
1570     my $blist_loc = $e->storage->read_at( $self->offset + $self->base_size, $e->byte_size );
1571     return unpack( $StP{$e->byte_size}, $blist_loc );
1572 }
1573
1574 sub get_bucket_list {
1575     my $self = shift;
1576     my ($args) = @_;
1577     $args ||= {};
1578
1579     # XXX Add in check here for recycling?
1580
1581     my $engine = $self->engine;
1582
1583     my $blist_loc = $self->get_blist_loc;
1584
1585     # There's no index or blist yet
1586     unless ( $blist_loc ) {
1587         return unless $args->{create};
1588
1589         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1590             engine  => $engine,
1591             key_md5 => $args->{key_md5},
1592         });
1593
1594         $engine->storage->print_at( $self->offset + $self->base_size,
1595             pack( $StP{$engine->byte_size}, $blist->offset ),
1596         );
1597
1598         return $blist;
1599     }
1600
1601     my $sector = $engine->_load_sector( $blist_loc )
1602         or DBM::Deep->_throw_error( "Cannot read sector at $blist_loc in get_bucket_list()" );
1603     my $i = 0;
1604     my $last_sector = undef;
1605     while ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
1606         $blist_loc = $sector->get_entry( ord( substr( $args->{key_md5}, $i++, 1 ) ) );
1607         $last_sector = $sector;
1608         if ( $blist_loc ) {
1609             $sector = $engine->_load_sector( $blist_loc )
1610                 or DBM::Deep->_throw_error( "Cannot read sector at $blist_loc in get_bucket_list()" );
1611         }
1612         else {
1613             $sector = undef;
1614             last;
1615         }
1616     }
1617
1618     # This means we went through the Index sector(s) and found an empty slot
1619     unless ( $sector ) {
1620         return unless $args->{create};
1621
1622         DBM::Deep->_throw_error( "No last_sector when attempting to build a new entry" )
1623             unless $last_sector;
1624
1625         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1626             engine  => $engine,
1627             key_md5 => $args->{key_md5},
1628         });
1629
1630         $last_sector->set_entry( ord( substr( $args->{key_md5}, $i - 1, 1 ) ) => $blist->offset );
1631
1632         return $blist;
1633     }
1634
1635     $sector->find_md5( $args->{key_md5} );
1636
1637     # See whether or not we need to reindex the bucketlist
1638     # Yes, the double-braces are there for a reason. if() doesn't create a redo-able block,
1639     # so we have to create a bare block within the if() for redo-purposes. Patch and idea
1640     # submitted by sprout@cpan.org. -RobK, 2008-01-09
1641     if ( !$sector->has_md5 && $args->{create} && $sector->{idx} == -1 ) {{
1642         my $redo;
1643
1644         my $new_index = DBM::Deep::Engine::Sector::Index->new({
1645             engine => $engine,
1646         });
1647
1648         my %blist_cache;
1649         #XXX q.v. the comments for this function.
1650         foreach my $entry ( $sector->chopped_up ) {
1651             my ($spot, $md5) = @{$entry};
1652             my $idx = ord( substr( $md5, $i, 1 ) );
1653
1654             # XXX This is inefficient
1655             my $blist = $blist_cache{$idx}
1656                 ||= DBM::Deep::Engine::Sector::BucketList->new({
1657                     engine => $engine,
1658                 });
1659
1660             $new_index->set_entry( $idx => $blist->offset );
1661
1662             my $new_spot = $blist->write_at_next_open( $md5 );
1663             $engine->reindex_entry( $spot => $new_spot );
1664         }
1665
1666         # Handle the new item separately.
1667         {
1668             my $idx = ord( substr( $args->{key_md5}, $i, 1 ) );
1669
1670             # If all the previous blist's items have been thrown into one
1671             # blist and the new item belongs in there too, we need
1672             # another index.
1673             if ( keys %blist_cache == 1 and each %blist_cache == $idx ) {
1674                 ++$i, ++$redo;
1675             } else {
1676                 my $blist = $blist_cache{$idx}
1677                     ||= DBM::Deep::Engine::Sector::BucketList->new({
1678                         engine => $engine,
1679                     });
1680
1681                 $new_index->set_entry( $idx => $blist->offset );
1682
1683                 #XXX THIS IS HACKY!
1684                 $blist->find_md5( $args->{key_md5} );
1685                 $blist->write_md5({
1686                     key     => $args->{key},
1687                     key_md5 => $args->{key_md5},
1688                     value   => DBM::Deep::Engine::Sector::Null->new({
1689                         engine => $engine,
1690                         data   => undef,
1691                     }),
1692                 });
1693             }
1694 #            my $blist = $blist_cache{$idx}
1695 #                ||= DBM::Deep::Engine::Sector::BucketList->new({
1696 #                    engine => $engine,
1697 #                });
1698 #
1699 #            $new_index->set_entry( $idx => $blist->offset );
1700 #
1701 #            #XXX THIS IS HACKY!
1702 #            $blist->find_md5( $args->{key_md5} );
1703 #            $blist->write_md5({
1704 #                key     => $args->{key},
1705 #                key_md5 => $args->{key_md5},
1706 #                value   => DBM::Deep::Engine::Sector::Null->new({
1707 #                    engine => $engine,
1708 #                    data   => undef,
1709 #                }),
1710 #            });
1711         }
1712
1713         if ( $last_sector ) {
1714             $last_sector->set_entry(
1715                 ord( substr( $args->{key_md5}, $i - 1, 1 ) ),
1716                 $new_index->offset,
1717             );
1718         } else {
1719             $engine->storage->print_at( $self->offset + $self->base_size,
1720                 pack( $StP{$engine->byte_size}, $new_index->offset ),
1721             );
1722         }
1723
1724         $sector->clear;
1725         $sector->free;
1726
1727         if ( $redo ) {
1728             (undef, $sector) = %blist_cache;
1729             $last_sector = $new_index;
1730             redo;
1731         }
1732
1733         $sector = $blist_cache{ ord( substr( $args->{key_md5}, $i, 1 ) ) };
1734         $sector->find_md5( $args->{key_md5} );
1735     }}
1736
1737     return $sector;
1738 }
1739
1740 sub get_class_offset {
1741     my $self = shift;
1742
1743     my $e = $self->engine;
1744     return unpack(
1745         $StP{$e->byte_size},
1746         $e->storage->read_at(
1747             $self->offset + $self->base_size + 1 * $e->byte_size, $e->byte_size,
1748         ),
1749     );
1750 }
1751
1752 sub get_classname {
1753     my $self = shift;
1754
1755     my $class_offset = $self->get_class_offset;
1756
1757     return unless $class_offset;
1758
1759     return $self->engine->_load_sector( $class_offset )->data;
1760 }
1761
1762 sub data {
1763     my $self = shift;
1764     my ($args) = @_;
1765     $args ||= {};
1766
1767     my $obj;
1768     unless ( $obj = $self->engine->cache->{ $self->offset } ) {
1769         $obj = DBM::Deep->new({
1770             type        => $self->type,
1771             base_offset => $self->offset,
1772             staleness   => $self->staleness,
1773             storage     => $self->engine->storage,
1774             engine      => $self->engine,
1775         });
1776
1777         if ( $self->engine->storage->{autobless} ) {
1778             my $classname = $self->get_classname;
1779             if ( defined $classname ) {
1780                 bless $obj, $classname;
1781             }
1782         }
1783
1784         $self->engine->cache->{$self->offset} = $obj;
1785     }
1786
1787     # We're not exporting, so just return.
1788     unless ( $args->{export} ) {
1789         return $obj;
1790     }
1791
1792     # We shouldn't export if this is still referred to.
1793     if ( $self->get_refcount > 1 ) {
1794         return $obj;
1795     }
1796
1797     return $obj->export;
1798 }
1799
1800 sub free {
1801     my $self = shift;
1802
1803     # We're not ready to be removed yet.
1804     if ( $self->decrement_refcount > 0 ) {
1805         return;
1806     }
1807
1808     # Rebless the object into DBM::Deep::Null.
1809     eval { %{ $self->engine->cache->{ $self->offset } } = (); };
1810     eval { @{ $self->engine->cache->{ $self->offset } } = (); };
1811     bless $self->engine->cache->{ $self->offset }, 'DBM::Deep::Null';
1812     delete $self->engine->cache->{ $self->offset };
1813
1814     my $blist_loc = $self->get_blist_loc;
1815     $self->engine->_load_sector( $blist_loc )->free if $blist_loc;
1816
1817     my $class_loc = $self->get_class_offset;
1818     $self->engine->_load_sector( $class_loc )->free if $class_loc;
1819
1820     $self->SUPER::free();
1821 }
1822
1823 sub increment_refcount {
1824     my $self = shift;
1825
1826     my $refcount = $self->get_refcount;
1827
1828     $refcount++;
1829
1830     $self->write_refcount( $refcount );
1831
1832     return $refcount;
1833 }
1834
1835 sub decrement_refcount {
1836     my $self = shift;
1837
1838     my $refcount = $self->get_refcount;
1839
1840     $refcount--;
1841
1842     $self->write_refcount( $refcount );
1843
1844     return $refcount;
1845 }
1846
1847 sub get_refcount {
1848     my $self = shift;
1849
1850     my $e = $self->engine;
1851     return unpack(
1852         $StP{$e->byte_size},
1853         $e->storage->read_at(
1854             $self->offset + $self->base_size + 2 * $e->byte_size, $e->byte_size,
1855         ),
1856     );
1857 }
1858
1859 sub write_refcount {
1860     my $self = shift;
1861     my ($num) = @_;
1862
1863     my $e = $self->engine;
1864     $e->storage->print_at(
1865         $self->offset + $self->base_size + 2 * $e->byte_size,
1866         pack( $StP{$e->byte_size}, $num ),
1867     );
1868 }
1869
1870 package DBM::Deep::Engine::Sector::BucketList;
1871
1872 our @ISA = qw( DBM::Deep::Engine::Sector );
1873
1874 sub _init {
1875     my $self = shift;
1876
1877     my $engine = $self->engine;
1878
1879     unless ( $self->offset ) {
1880         my $leftover = $self->size - $self->base_size;
1881
1882         $self->{offset} = $engine->_request_blist_sector( $self->size );
1883         $engine->storage->print_at( $self->offset, $engine->SIG_BLIST ); # Sector type
1884         # Skip staleness counter
1885         $engine->storage->print_at( $self->offset + $self->base_size,
1886             chr(0) x $leftover, # Zero-fill the data
1887         );
1888     }
1889
1890     if ( $self->{key_md5} ) {
1891         $self->find_md5;
1892     }
1893
1894     return $self;
1895 }
1896
1897 sub clear {
1898     my $self = shift;
1899     $self->engine->storage->print_at( $self->offset + $self->base_size,
1900         chr(0) x ($self->size - $self->base_size), # Zero-fill the data
1901     );
1902 }
1903
1904 sub size {
1905     my $self = shift;
1906     unless ( $self->{size} ) {
1907         my $e = $self->engine;
1908         # Base + numbuckets * bucketsize
1909         $self->{size} = $self->base_size + $e->max_buckets * $self->bucket_size;
1910     }
1911     return $self->{size};
1912 }
1913
1914 sub free_meth { return '_add_free_blist_sector' }
1915
1916 sub free {
1917     my $self = shift;
1918
1919     my $e = $self->engine;
1920     foreach my $bucket ( $self->chopped_up ) {
1921         my $rest = $bucket->[-1];
1922
1923         # Delete the keysector
1924         my $l = unpack( $StP{$e->byte_size}, substr( $rest, $e->hash_size, $e->byte_size ) );
1925         my $s = $e->_load_sector( $l ); $s->free if $s;
1926
1927         # Delete the HEAD sector
1928         $l = unpack( $StP{$e->byte_size},
1929             substr( $rest,
1930                 $e->hash_size + $e->byte_size,
1931                 $e->byte_size,
1932             ),
1933         );
1934         $s = $e->_load_sector( $l ); $s->free if $s;
1935
1936         foreach my $txn ( 0 .. $e->num_txns - 2 ) {
1937             my $l = unpack( $StP{$e->byte_size},
1938                 substr( $rest,
1939                     $e->hash_size + 2 * $e->byte_size + $txn * ($e->byte_size + $STALE_SIZE),
1940                     $e->byte_size,
1941                 ),
1942             );
1943             my $s = $e->_load_sector( $l ); $s->free if $s;
1944         }
1945     }
1946
1947     $self->SUPER::free();
1948 }
1949
1950 sub bucket_size {
1951     my $self = shift;
1952     unless ( $self->{bucket_size} ) {
1953         my $e = $self->engine;
1954         # Key + head (location) + transactions (location + staleness-counter)
1955         my $location_size = $e->byte_size + $e->byte_size + ($e->num_txns - 1) * ($e->byte_size + $STALE_SIZE);
1956         $self->{bucket_size} = $e->hash_size + $location_size;
1957     }
1958     return $self->{bucket_size};
1959 }
1960
1961 # XXX This is such a poor hack. I need to rethink this code.
1962 sub chopped_up {
1963     my $self = shift;
1964
1965     my $e = $self->engine;
1966
1967     my @buckets;
1968     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1969         my $spot = $self->offset + $self->base_size + $idx * $self->bucket_size;
1970         my $md5 = $e->storage->read_at( $spot, $e->hash_size );
1971
1972         #XXX If we're chopping, why would we ever have the blank_md5?
1973         last if $md5 eq $e->blank_md5;
1974
1975         my $rest = $e->storage->read_at( undef, $self->bucket_size - $e->hash_size );
1976         push @buckets, [ $spot, $md5 . $rest ];
1977     }
1978
1979     return @buckets;
1980 }
1981
1982 sub write_at_next_open {
1983     my $self = shift;
1984     my ($entry) = @_;
1985
1986     #XXX This is such a hack!
1987     $self->{_next_open} = 0 unless exists $self->{_next_open};
1988
1989     my $spot = $self->offset + $self->base_size + $self->{_next_open}++ * $self->bucket_size;
1990     $self->engine->storage->print_at( $spot, $entry );
1991
1992     return $spot;
1993 }
1994
1995 sub has_md5 {
1996     my $self = shift;
1997     unless ( exists $self->{found} ) {
1998         $self->find_md5;
1999     }
2000     return $self->{found};
2001 }
2002
2003 sub find_md5 {
2004     my $self = shift;
2005
2006     $self->{found} = undef;
2007     $self->{idx}   = -1;
2008
2009     if ( @_ ) {
2010         $self->{key_md5} = shift;
2011     }
2012
2013     # If we don't have an MD5, then what are we supposed to do?
2014     unless ( exists $self->{key_md5} ) {
2015         DBM::Deep->_throw_error( "Cannot find_md5 without a key_md5 set" );
2016     }
2017
2018     my $e = $self->engine;
2019     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
2020         my $potential = $e->storage->read_at(
2021             $self->offset + $self->base_size + $idx * $self->bucket_size, $e->hash_size,
2022         );
2023
2024         if ( $potential eq $e->blank_md5 ) {
2025             $self->{idx} = $idx;
2026             return;
2027         }
2028
2029         if ( $potential eq $self->{key_md5} ) {
2030             $self->{found} = 1;
2031             $self->{idx} = $idx;
2032             return;
2033         }
2034     }
2035
2036     return;
2037 }
2038
2039 sub write_md5 {
2040     my $self = shift;
2041     my ($args) = @_;
2042
2043     DBM::Deep->_throw_error( "write_md5: no key" ) unless exists $args->{key};
2044     DBM::Deep->_throw_error( "write_md5: no key_md5" ) unless exists $args->{key_md5};
2045     DBM::Deep->_throw_error( "write_md5: no value" ) unless exists $args->{value};
2046
2047     my $engine = $self->engine;
2048
2049     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
2050
2051     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2052     $engine->add_entry( $args->{trans_id}, $spot );
2053
2054     unless ($self->{found}) {
2055         my $key_sector = DBM::Deep::Engine::Sector::Scalar->new({
2056             engine => $engine,
2057             data   => $args->{key},
2058         });
2059
2060         $engine->storage->print_at( $spot,
2061             $args->{key_md5},
2062             pack( $StP{$engine->byte_size}, $key_sector->offset ),
2063         );
2064     }
2065
2066     my $loc = $spot
2067       + $engine->hash_size
2068       + $engine->byte_size;
2069
2070     if ( $args->{trans_id} ) {
2071         $loc += $engine->byte_size + ($args->{trans_id} - 1) * ( $engine->byte_size + $STALE_SIZE );
2072
2073         $engine->storage->print_at( $loc,
2074             pack( $StP{$engine->byte_size}, $args->{value}->offset ),
2075             pack( $StP{$STALE_SIZE}, $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
2076         );
2077     }
2078     else {
2079         $engine->storage->print_at( $loc,
2080             pack( $StP{$engine->byte_size}, $args->{value}->offset ),
2081         );
2082     }
2083 }
2084
2085 sub mark_deleted {
2086     my $self = shift;
2087     my ($args) = @_;
2088     $args ||= {};
2089
2090     my $engine = $self->engine;
2091
2092     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
2093
2094     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2095     $engine->add_entry( $args->{trans_id}, $spot );
2096
2097     my $loc = $spot
2098       + $engine->hash_size
2099       + $engine->byte_size;
2100
2101     if ( $args->{trans_id} ) {
2102         $loc += $engine->byte_size + ($args->{trans_id} - 1) * ( $engine->byte_size + $STALE_SIZE );
2103
2104         $engine->storage->print_at( $loc,
2105             pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
2106             pack( $StP{$STALE_SIZE}, $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
2107         );
2108     }
2109     else {
2110         $engine->storage->print_at( $loc,
2111             pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
2112         );
2113     }
2114
2115 }
2116
2117 sub delete_md5 {
2118     my $self = shift;
2119     my ($args) = @_;
2120
2121     my $engine = $self->engine;
2122     return undef unless $self->{found};
2123
2124     # Save the location so that we can free the data
2125     my $location = $self->get_data_location_for({
2126         allow_head => 0,
2127     });
2128     my $key_sector = $self->get_key_for;
2129
2130     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
2131     $engine->storage->print_at( $spot,
2132         $engine->storage->read_at(
2133             $spot + $self->bucket_size,
2134             $self->bucket_size * ( $engine->max_buckets - $self->{idx} - 1 ),
2135         ),
2136         chr(0) x $self->bucket_size,
2137     );
2138
2139     $key_sector->free;
2140
2141     my $data_sector = $self->engine->_load_sector( $location );
2142     my $data = $data_sector->data({ export => 1 });
2143     $data_sector->free;
2144
2145     return $data;
2146 }
2147
2148 sub get_data_location_for {
2149     my $self = shift;
2150     my ($args) = @_;
2151     $args ||= {};
2152
2153     $args->{allow_head} = 0 unless exists $args->{allow_head};
2154     $args->{trans_id}   = $self->engine->trans_id unless exists $args->{trans_id};
2155     $args->{idx}        = $self->{idx} unless exists $args->{idx};
2156
2157     my $e = $self->engine;
2158
2159     my $spot = $self->offset + $self->base_size
2160       + $args->{idx} * $self->bucket_size
2161       + $e->hash_size
2162       + $e->byte_size;
2163
2164     if ( $args->{trans_id} ) {
2165         $spot += $e->byte_size + ($args->{trans_id} - 1) * ( $e->byte_size + $STALE_SIZE );
2166     }
2167
2168     my $buffer = $e->storage->read_at(
2169         $spot,
2170         $e->byte_size + $STALE_SIZE,
2171     );
2172     my ($loc, $staleness) = unpack( $StP{$e->byte_size} . ' ' . $StP{$STALE_SIZE}, $buffer );
2173
2174     # XXX Merge the two if-clauses below
2175     if ( $args->{trans_id} ) {
2176         # We have found an entry that is old, so get rid of it
2177         if ( $staleness != (my $s = $e->get_txn_staleness_counter( $args->{trans_id} ) ) ) {
2178             $e->storage->print_at(
2179                 $spot,
2180                 pack( $StP{$e->byte_size} . ' ' . $StP{$STALE_SIZE}, (0) x 2 ),
2181             );
2182             $loc = 0;
2183         }
2184     }
2185
2186     # If we're in a transaction and we never wrote to this location, try the
2187     # HEAD instead.
2188     if ( $args->{trans_id} && !$loc && $args->{allow_head} ) {
2189         return $self->get_data_location_for({
2190             trans_id   => 0,
2191             allow_head => 1,
2192             idx        => $args->{idx},
2193         });
2194     }
2195
2196     return $loc <= 1 ? 0 : $loc;
2197 }
2198
2199 sub get_data_for {
2200     my $self = shift;
2201     my ($args) = @_;
2202     $args ||= {};
2203
2204     return unless $self->{found};
2205     my $location = $self->get_data_location_for({
2206         allow_head => $args->{allow_head},
2207     });
2208     return $self->engine->_load_sector( $location );
2209 }
2210
2211 sub get_key_for {
2212     my $self = shift;
2213     my ($idx) = @_;
2214     $idx = $self->{idx} unless defined $idx;
2215
2216     if ( $idx >= $self->engine->max_buckets ) {
2217         DBM::Deep->_throw_error( "get_key_for(): Attempting to retrieve $idx" );
2218     }
2219
2220     my $location = $self->engine->storage->read_at(
2221         $self->offset + $self->base_size + $idx * $self->bucket_size + $self->engine->hash_size,
2222         $self->engine->byte_size,
2223     );
2224     $location = unpack( $StP{$self->engine->byte_size}, $location );
2225     DBM::Deep->_throw_error( "get_key_for: No location?" ) unless $location;
2226
2227     return $self->engine->_load_sector( $location );
2228 }
2229
2230 package DBM::Deep::Engine::Sector::Index;
2231
2232 our @ISA = qw( DBM::Deep::Engine::Sector );
2233
2234 sub _init {
2235     my $self = shift;
2236
2237     my $engine = $self->engine;
2238
2239     unless ( $self->offset ) {
2240         my $leftover = $self->size - $self->base_size;
2241
2242         $self->{offset} = $engine->_request_index_sector( $self->size );
2243         $engine->storage->print_at( $self->offset, $engine->SIG_INDEX ); # Sector type
2244         # Skip staleness counter
2245         $engine->storage->print_at( $self->offset + $self->base_size,
2246             chr(0) x $leftover, # Zero-fill the rest
2247         );
2248     }
2249
2250     return $self;
2251 }
2252
2253 #XXX Change here
2254 sub size {
2255     my $self = shift;
2256     unless ( $self->{size} ) {
2257         my $e = $self->engine;
2258         $self->{size} = $self->base_size + $e->byte_size * $e->hash_chars;
2259     }
2260     return $self->{size};
2261 }
2262
2263 sub free_meth { return '_add_free_index_sector' }
2264
2265 sub free {
2266     my $self = shift;
2267     my $e = $self->engine;
2268
2269     for my $i ( 0 .. $e->hash_chars - 1 ) {
2270         my $l = $self->get_entry( $i ) or next;
2271         $e->_load_sector( $l )->free;
2272     }
2273
2274     $self->SUPER::free();
2275 }
2276
2277 sub _loc_for {
2278     my $self = shift;
2279     my ($idx) = @_;
2280     return $self->offset + $self->base_size + $idx * $self->engine->byte_size;
2281 }
2282
2283 sub get_entry {
2284     my $self = shift;
2285     my ($idx) = @_;
2286
2287     my $e = $self->engine;
2288
2289     DBM::Deep->_throw_error( "get_entry: Out of range ($idx)" )
2290         if $idx < 0 || $idx >= $e->hash_chars;
2291
2292     return unpack(
2293         $StP{$e->byte_size},
2294         $e->storage->read_at( $self->_loc_for( $idx ), $e->byte_size ),
2295     );
2296 }
2297
2298 sub set_entry {
2299     my $self = shift;
2300     my ($idx, $loc) = @_;
2301
2302     my $e = $self->engine;
2303
2304     DBM::Deep->_throw_error( "set_entry: Out of range ($idx)" )
2305         if $idx < 0 || $idx >= $e->hash_chars;
2306
2307     $self->engine->storage->print_at(
2308         $self->_loc_for( $idx ),
2309         pack( $StP{$e->byte_size}, $loc ),
2310     );
2311 }
2312
2313 # This was copied from MARCEL's Class::Null. However, I couldn't use it because
2314 # I need an undef value, not an implementation of the Null Class pattern.
2315 package DBM::Deep::Null;
2316
2317 use overload
2318     'bool'   => sub { undef },
2319     '""'     => sub { undef },
2320     '0+'     => sub { undef },
2321     fallback => 1,
2322     nomethod => 'AUTOLOAD';
2323
2324 sub AUTOLOAD { return; }
2325
2326 1;
2327 __END__