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